Asterisk
/
asterisk
mirror of https://github.com/asterisk/asterisk
1
0
Fork 0
Code Issues Releases Wiki Activity

stasis cache: Enhance to keep track of an item from different entities.

Browse Source 
A stasis cache entry now contains more than a single message/snapshot.  It
contains messages/snapshots for the local entity as well as any remote
entities that post to the cached item.  In addition callbacks can be
supplied when the cache is created to compute and post the aggregate
message/snapshot representing all entities stored in the cache entry.

* All stasis messages now have an eid to indicate what entity posted it.

* The stasis cache enhancements allow device state to cache and aggregate
the device states from local and remote entities in a single operation.
The cached aggregate device state is available immediately after it is
posted to the stasis bus.  This improves performance by eliminating a
cache dump and associated ao2 container traversals to calculate the
aggregate state.

(closes issue ASTERISK-23204)
Reported by: Mark Michelson

Review: https://reviewboard.asterisk.org/r/3281/
........

Merged revisions 410184 from http://svn.asterisk.org/svn/asterisk/branches/12


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@410185 65c4cc65-6c06-0410-ace0-fbb531ad65f3
changes/97/197/1
Richard Mudgett 11 years ago
parent ecbd052741
commit 4ad1245cb5
8 changed files with 1459 additions and 379 deletions
Show Stats Download Patch File Download Diff File

21
include/asterisk/devicestate.h
Unescape View File

@ -274,13 +274,20 @@ struct ast_devstate_aggregate {
* \since 12
*/
struct ast_device_state_message {
AST_DECLARE_STRING_FIELDS(
AST_STRING_FIELD(cache_id); /*!< A unique ID used for hashing */
AST_STRING_FIELD(device); /*!< The name of the device */
);
enum ast_device_state state; /*!< The state of the device */
struct ast_eid *eid; /*!< The EID of the server where this message originated, NULL EID means aggregate state */
enum ast_devstate_cache cachable; /*!< Flag designating the cachability of this device state */
/*! The name of the device */
const char *device;
/*!
* \brief The EID of the server where this message originated.
*
* \note A NULL EID means aggregate state.
*/
const struct ast_eid *eid;
/*! The state of the device */
enum ast_device_state state;
/*! Flag designating the cachability of this device state */
enum ast_devstate_cache cachable;
/*! The device and eid data is stuffed here when the struct is allocated. */
struct ast_eid stuff[0];
};
/*!

258
include/asterisk/stasis.h
Unescape View File

@ -290,12 +290,50 @@ const char *stasis_message_type_name(const struct stasis_message_type *type);
*
* \param type Type of the message
* \param data Immutable data that is the actual contents of the message
*
* \return New message
* \return \c NULL on error
*
* \since 12
*/
struct stasis_message *stasis_message_create(struct stasis_message_type *type, void *data);
/*!
* \brief Create a new message for an entity.
*
* This message is an \c ao2 object, and must be ao2_cleanup()'ed when you are done
* with it. Messages are also immutable, and must not be modified after they
* are initialized. Especially the \a data in the message.
*
* \param type Type of the message
* \param data Immutable data that is the actual contents of the message
* \param eid What entity originated this message. (NULL for aggregate)
*
* \note An aggregate message is a combined representation of the local
* and remote entities publishing the message data. e.g., An aggregate
* device state represents the combined device state from the local and
* any remote entities publishing state for a device. e.g., An aggregate
* MWI message is the old/new MWI counts accumulated from the local and
* any remote entities publishing to a mailbox.
*
* \retval New message
* \retval \c NULL on error
*
* \since 12.2.0
*/
struct stasis_message *stasis_message_create_full(struct stasis_message_type *type, void *data, const struct ast_eid *eid);
/*!
* \brief Get the entity id for a \ref stasis_message.
* \since 12.2.0
*
* \param msg Message to get eid.
*
* \retval Entity id of \a msg
* \retval \c NULL if \a msg is an aggregate or \a msg is \c NULL.
*/
const struct ast_eid *stasis_message_eid(const struct stasis_message *msg);
/*!
* \brief Get the message type for a \ref stasis_message.
* \param msg Message to type
@ -503,8 +541,8 @@ struct stasis_forward;
* \brief Create a subscription which forwards all messages from one topic to
* another.
*
* Note that the \a topic parameter of the invoked callback will the be \a topic
* the message was sent to, not the topic the subscriber subscribed to.
* Note that the \a topic parameter of the invoked callback will the be the
* \a topic the message was sent to, not the topic the subscriber subscribed to.
*
* \param from_topic Topic to forward.
* \param to_topic Destination topic of forwarded messages.
@ -640,6 +678,9 @@ struct stasis_message_type *stasis_cache_clear_type(void);
*/
struct stasis_cache;
/*! Cache entry used for calculating the aggregate snapshot. */
struct stasis_cache_entry;
/*!
* \brief A topic wrapper, which caches certain messages.
* \since 12
@ -660,6 +701,101 @@ struct stasis_caching_topic;
*/
typedef const char *(*snapshot_get_id)(struct stasis_message *message);
/*!
* \brief Callback to calculate the aggregate cache entry.
* \since 12.2.0
*
* \param entry Cache entry to calculate a new aggregate snapshot.
* \param new_snapshot The shapshot that is being updated.
*
* \note Return a ref bumped pointer from stasis_cache_entry_get_aggregate()
* if a new aggregate could not be calculated because of error.
*
* \note An aggregate message is a combined representation of the local
* and remote entities publishing the message data. e.g., An aggregate
* device state represents the combined device state from the local and
* any remote entities publishing state for a device. e.g., An aggregate
* MWI message is the old/new MWI counts accumulated from the local and
* any remote entities publishing to a mailbox.
*
* \return New aggregate-snapshot calculated on success.
* Caller has a reference on return.
*/
typedef struct stasis_message *(*cache_aggregate_calc_fn)(struct stasis_cache_entry *entry, struct stasis_message *new_snapshot);
/*!
* \brief Callback to publish the aggregate cache entry message.
* \since 12.2.0
*
* \details
* Once an aggregate message is calculated. This callback publishes the
* message so subscribers will know the new value of an aggregated state.
*
* \param topic The aggregate message may be published to this topic.
* It is the topic to which the cache itself is subscribed.
* \param aggregate The aggregate shapshot message to publish.
*
* \note It is up to the function to determine if there is a better topic
* the aggregate message should be published over.
*
* \note An aggregate message is a combined representation of the local
* and remote entities publishing the message data. e.g., An aggregate
* device state represents the combined device state from the local and
* any remote entities publishing state for a device. e.g., An aggregate
* MWI message is the old/new MWI counts accumulated from the local and
* any remote entities publishing to a mailbox.
*
* \return Nothing
*/
typedef void (*cache_aggregate_publish_fn)(struct stasis_topic *topic, struct stasis_message *aggregate);
/*!
* \brief Get the aggregate cache entry snapshot.
* \since 12.2.0
*
* \param entry Cache entry to get the aggregate snapshot.
*
* \note A reference is not given to the returned pointer so don't unref it.
*
* \note An aggregate message is a combined representation of the local
* and remote entities publishing the message data. e.g., An aggregate
* device state represents the combined device state from the local and
* any remote entities publishing state for a device. e.g., An aggregate
* MWI message is the old/new MWI counts accumulated from the local and
* any remote entities publishing to a mailbox.
*
* \retval Aggregate-snapshot in cache.
* \retval NULL if not present.
*/
struct stasis_message *stasis_cache_entry_get_aggregate(struct stasis_cache_entry *entry);
/*!
* \brief Get the local entity's cache entry snapshot.
* \since 12.2.0
*
* \param entry Cache entry to get the local entity's snapshot.
*
* \note A reference is not given to the returned pointer so don't unref it.
*
* \retval Internal-snapshot in cache.
* \retval NULL if not present.
*/
struct stasis_message *stasis_cache_entry_get_local(struct stasis_cache_entry *entry);
/*!
* \brief Get a remote entity's cache entry snapshot by index.
* \since 12.2.0
*
* \param entry Cache entry to get a remote entity's snapshot.
* \param idx Which remote entity's snapshot to get.
*
* \note A reference is not given to the returned pointer so don't unref it.
*
* \retval Remote-entity-snapshot in cache.
* \retval NULL if not present.
*/
struct stasis_message *stasis_cache_entry_get_remote(struct stasis_cache_entry *entry, int idx);
/*!
* \brief Create a cache.
*
@ -669,12 +805,40 @@ typedef const char *(*snapshot_get_id)(struct stasis_message *message);
* The returned object is AO2 managed, so ao2_cleanup() when you're done.
*
* \param id_fn Callback to extract the id from a snapshot message.
* \return New cache indexed by \a id_fn.
* \return \c NULL on error
*
* \retval New cache indexed by \a id_fn.
* \retval \c NULL on error
*
* \since 12
*/
struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn);
/*!
* \brief Create a cache.
*
* This is the backend store for a \ref stasis_caching_topic. The cache is
* thread safe, allowing concurrent reads and writes.
*
* The returned object is AO2 managed, so ao2_cleanup() when you're done.
*
* \param id_fn Callback to extract the id from a snapshot message.
* \param aggregate_calc_fn Callback to calculate the aggregate cache entry.
* \param aggregate_publish_fn Callback to publish the aggregate cache entry.
*
* \note An aggregate message is a combined representation of the local
* and remote entities publishing the message data. e.g., An aggregate
* device state represents the combined device state from the local and
* any remote entities publishing state for a device. e.g., An aggregate
* MWI message is the old/new MWI counts accumulated from the local and
* any remote entities publishing to a mailbox.
*
* \retval New cache indexed by \a id_fn.
* \retval \c NULL on error
*
* \since 12.2.0
*/
struct stasis_cache *stasis_cache_create_full(snapshot_get_id id_fn, cache_aggregate_calc_fn aggregate_calc_fn, cache_aggregate_publish_fn aggregate_publish_fn);
/*!
* \brief Create a topic which monitors and caches messages from another topic.
*
@ -749,31 +913,95 @@ struct stasis_topic *stasis_caching_get_topic(
struct stasis_message *stasis_cache_clear_create(struct stasis_message *message);
/*!
* \brief Retrieve an item from the cache.
* \brief Retrieve an item from the cache for the ast_eid_default entity.
*
* The returned item is AO2 managed, so ao2_cleanup() when you're done with it.
*
* \param cache The cache to query.
* \param type Type of message to retrieve.
* \param id Identity of the snapshot to retrieve.
* \return Message from the cache.
* \return \c NULL if message is not found.
*
* \retval Message from the cache.
* \retval \c NULL if message is not found.
*
* \since 12
*/
struct stasis_message *stasis_cache_get(
struct stasis_cache *cache, struct stasis_message_type *type,
const char *id);
struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id);
/*!
* \brief Dump cached items to a subscription
* \brief Retrieve an item from the cache for a specific entity.
*
* The returned item is AO2 managed, so ao2_cleanup() when you're done with it.
*
* \param cache The cache to query.
* \param type Type of message to retrieve.
* \param id Identity of the snapshot to retrieve.
* \param eid Specific entity id to retrieve. NULL for aggregate.
*
* \note An aggregate message is a combined representation of the local
* and remote entities publishing the message data. e.g., An aggregate
* device state represents the combined device state from the local and
* any remote entities publishing state for a device. e.g., An aggregate
* MWI message is the old/new MWI counts accumulated from the local and
* any remote entities publishing to a mailbox.
*
* \retval Message from the cache.
* \retval \c NULL if message is not found.
*
* \since 12.2.0
*/
struct stasis_message *stasis_cache_get_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const char *id, const struct ast_eid *eid);
/*!
* \brief Retrieve all matching entity items from the cache.
* \since 12.2.0
*
* \param cache The cache to query.
* \param type Type of message to retrieve.
* \param id Identity of the snapshot to retrieve.
*
* \retval Container of matching items found.
* \retval \c NULL if error.
*/
struct ao2_container *stasis_cache_get_all(struct stasis_cache *cache, struct stasis_message_type *type, const char *id);
/*!
* \brief Dump cached items to a subscription for the ast_eid_default entity.
*
* \param cache The cache to query.
* \param type Type of message to dump (any type if \c NULL).
* \return ao2_container containing all matches (must be unreffed by caller)
* \return \c NULL on allocation error
*
* \retval ao2_container containing all matches (must be unreffed by caller)
* \retval \c NULL on allocation error
*
* \since 12
*/
struct ao2_container *stasis_cache_dump(struct stasis_cache *cache,
struct stasis_message_type *type);
struct ao2_container *stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type);
/*!
* \brief Dump cached items to a subscription for a specific entity.
* \since 12.2.0
*
* \param cache The cache to query.
* \param type Type of message to dump (any type if \c NULL).
* \param eid Specific entity id to retrieve. NULL for aggregate.
*
* \retval ao2_container containing all matches (must be unreffed by caller)
* \retval \c NULL on allocation error
*/
struct ao2_container *stasis_cache_dump_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid);
/*!
* \brief Dump all entity items from the cache to a subscription.
* \since 12.2.0
*
* \param cache The cache to query.
* \param type Type of message to dump (any type if \c NULL).
*
* \retval ao2_container containing all matches (must be unreffed by caller)
* \retval \c NULL on allocation error
*/
struct ao2_container *stasis_cache_dump_all(struct stasis_cache *cache, struct stasis_message_type *type);
/*! @} */

15
main/app.c
Unescape View File

@ -2821,7 +2821,6 @@ struct ast_mwi_state *ast_mwi_create(const char *mailbox, const char *context)
return mwi_state;
}
int ast_publish_mwi_state_full(
const char *mailbox,
const char *context,
@ -2857,10 +2856,19 @@ int ast_publish_mwi_state_full(
if (eid) {
mwi_state->eid = *eid;
} else {
ast_set_default_eid(&mwi_state->eid);
mwi_state->eid = ast_eid_default;
}
message = stasis_message_create(ast_mwi_state_type(), mwi_state);
/*
* As far as stasis is concerned, all MWI events are internal.
*
* We may in the future want to make MWI aggregate internal/external
* message counts similar to how device state aggregates state.
*/
message = stasis_message_create_full(ast_mwi_state_type(), mwi_state, &ast_eid_default);
if (!message) {
return -1;
}
mailbox_specific_topic = ast_mwi_topic(mwi_state->uniqueid);
if (!mailbox_specific_topic) {
@ -2911,6 +2919,7 @@ struct stasis_message *ast_mwi_blob_create(struct ast_mwi_state *mwi_state,
ao2_ref(obj->mwi_state, +1);
obj->blob = ast_json_ref(blob);
/* This is not a normal MWI event. Only used by the MinivmNotify app. */
msg = stasis_message_create(message_type, obj);
if (!msg) {
return NULL;

327
main/devicestate.c
Unescape View File

@ -282,16 +282,20 @@ enum ast_device_state ast_parse_device_state(const char *device)
static enum ast_device_state devstate_cached(const char *device)
{
RAII_VAR(struct stasis_message *, cached_msg, NULL, ao2_cleanup);
struct stasis_message *cached_msg;
struct ast_device_state_message *device_state;
enum ast_device_state state;
cached_msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), device);
cached_msg = stasis_cache_get_by_eid(ast_device_state_cache(),
ast_device_state_message_type(), device, NULL);
if (!cached_msg) {
return AST_DEVICE_UNKNOWN;
}
device_state = stasis_message_data(cached_msg);
state = device_state->state;
ao2_cleanup(cached_msg);
return device_state->state;
return state;
}
/*! \brief Check device state through channel specific function or generic function */
@ -522,148 +526,62 @@ static void *do_devstate_changes(void *data)
return NULL;
}
#define MAX_SERVERS 64
static int devstate_change_aggregator_cb(void *obj, void *arg, void *data, int flags)
{
struct stasis_message *msg = obj;
struct ast_devstate_aggregate *aggregate = arg;
char *device = data;
struct ast_device_state_message *device_state = stasis_message_data(msg);
if (!device_state->eid || strcmp(device, device_state->device)) {
/* ignore aggregate states and devices that don't match */
return 0;
}
ast_debug(1, "Adding per-server state of '%s' for '%s'\n",
ast_devstate2str(device_state->state), device);
ast_devstate_aggregate_add(aggregate, device_state->state);
return 0;
}
static void device_state_dtor(void *obj)
{
struct ast_device_state_message *device_state = obj;
ast_string_field_free_memory(device_state);
ast_free(device_state->eid);
}
static struct ast_device_state_message *device_state_alloc(const char *device, enum ast_device_state state, enum ast_devstate_cache cachable, const struct ast_eid *eid)
{
RAII_VAR(struct ast_device_state_message *, new_device_state, ao2_alloc(sizeof(*new_device_state), device_state_dtor), ao2_cleanup);
struct ast_device_state_message *new_device_state;
char *pos;
size_t stuff_len;
ast_assert(!ast_strlen_zero(device));
if (!new_device_state || ast_string_field_init(new_device_state, 256)) {
stuff_len = strlen(device) + 1;
if (eid) {
stuff_len += sizeof(*eid);
}
new_device_state = ao2_alloc_options(sizeof(*new_device_state) + stuff_len, NULL,
AO2_ALLOC_OPT_LOCK_NOLOCK);
if (!new_device_state) {
return NULL;
}
ast_string_field_set(new_device_state, device, device);
new_device_state->state = state;
new_device_state->cachable = cachable;
if (eid) {
char eid_str[20];
struct ast_str *cache_id = ast_str_alloca(256);
new_device_state->eid = ast_malloc(sizeof(*eid));
if (!new_device_state->eid) {
return NULL;
}
*new_device_state->eid = *eid;
ast_eid_to_str(eid_str, sizeof(eid_str), new_device_state->eid);
ast_str_set(&cache_id, 0, "%s%s", eid_str, device);
ast_string_field_set(new_device_state, cache_id, ast_str_buffer(cache_id));
/* non-aggregate device state. */
new_device_state->stuff[0] = *eid;
new_device_state->eid = &new_device_state->stuff[0];
pos = (char *) &new_device_state->stuff[1];
} else {
/* no EID makes this an aggregate state */
ast_string_field_set(new_device_state, cache_id, device);
pos = (char *) &new_device_state->stuff[0];
}
ao2_ref(new_device_state, +1);
return new_device_state;
}
static enum ast_device_state get_aggregate_state(char *device)
{
RAII_VAR(struct ao2_container *, cached, NULL, ao2_cleanup);
struct ast_devstate_aggregate aggregate;
ast_devstate_aggregate_init(&aggregate);
cached = stasis_cache_dump(ast_device_state_cache(), NULL);
ao2_callback_data(cached, OBJ_NODATA, devstate_change_aggregator_cb, &aggregate, device);
return ast_devstate_aggregate_result(&aggregate);
}
static int aggregate_state_changed(char *device, enum ast_device_state new_aggregate_state)
{
RAII_VAR(struct stasis_message *, cached_aggregate_msg, NULL, ao2_cleanup);
struct ast_device_state_message *cached_aggregate_device_state;
strcpy(pos, device);/* Safe */
new_device_state->device = pos;
cached_aggregate_msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), device);
if (!cached_aggregate_msg) {
return 1;
}
new_device_state->state = state;
new_device_state->cachable = cachable;
cached_aggregate_device_state = stasis_message_data(cached_aggregate_msg);
if (cached_aggregate_device_state->state == new_aggregate_state) {
return 0;
}
return 1;
return new_device_state;
}
static void devstate_change_collector_cb(void *data, struct stasis_subscription *sub, struct stasis_message *msg)
static void devstate_change_cb(void *data, struct stasis_subscription *sub, struct stasis_message *msg)
{
enum ast_device_state aggregate_state;
char *device;
struct ast_device_state_message *device_state;
RAII_VAR(struct stasis_message *, new_aggregate_msg, NULL, ao2_cleanup);
RAII_VAR(struct ast_device_state_message *, new_aggregate_state, NULL, ao2_cleanup);
if (stasis_cache_update_type() == stasis_message_type(msg)) {
struct stasis_cache_update *update = stasis_message_data(msg);
if (!update->new_snapshot) {
return;
}
msg = update->new_snapshot;
}
if (ast_device_state_message_type() != stasis_message_type(msg)) {
return;
}
device_state = stasis_message_data(msg);
if (!device_state->eid) {
/* ignore aggregate messages */
if (device_state->cachable == AST_DEVSTATE_CACHABLE || !device_state->eid) {
/* Ignore cacheable and aggregate messages. */
return;
}
device = ast_strdupa(device_state->device);
ast_debug(1, "Processing device state change for '%s'\n", device);
if (device_state->cachable == AST_DEVSTATE_NOT_CACHABLE) {
/* if it's not cachable, there will be no aggregate state to get
* and this should be passed through */
aggregate_state = device_state->state;
} else {
aggregate_state = get_aggregate_state(device);
ast_debug(1, "Aggregate devstate result is '%s' for '%s'\n",
ast_devstate2str(aggregate_state), device);
if (!aggregate_state_changed(device, aggregate_state)) {
/* No change since last reported device state */
ast_debug(1, "Aggregate state for device '%s' has not changed from '%s'\n",
device, ast_devstate2str(aggregate_state));
return;
}
}
ast_debug(1, "Aggregate state for device '%s' has changed to '%s'\n",
device, ast_devstate2str(aggregate_state));
ast_publish_device_state_full(device, aggregate_state, device_state->cachable, NULL);
/*
* Non-cacheable device state aggregates are just the
* device state republished as the aggregate.
*/
ast_publish_device_state_full(device_state->device, device_state->state,
device_state->cachable, NULL);
}
/*! \brief Initialize the device state engine in separate thread */
@ -738,25 +656,30 @@ struct stasis_topic *ast_device_state_topic(const char *device)
int ast_device_state_clear_cache(const char *device)
{
RAII_VAR(struct stasis_message *, cached_msg, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
struct stasis_message *cached_msg;
struct stasis_message *msg;
if (!(cached_msg = stasis_cache_get(ast_device_state_cache(),
ast_device_state_message_type(), device))) {
cached_msg = stasis_cache_get_by_eid(ast_device_state_cache(),
ast_device_state_message_type(), device, &ast_eid_default);
if (!cached_msg) {
/* nothing to clear */
return -1;
}
msg = stasis_cache_clear_create(cached_msg);
stasis_publish(ast_device_state_topic(device), msg);
if (msg) {
stasis_publish(ast_device_state_topic(device), msg);
}
ao2_cleanup(msg);
ao2_cleanup(cached_msg);
return 0;
}
int ast_publish_device_state_full(
const char *device,
enum ast_device_state state,
enum ast_devstate_cache cachable,
struct ast_eid *eid)
const char *device,
enum ast_device_state state,
enum ast_devstate_cache cachable,
struct ast_eid *eid)
{
RAII_VAR(struct ast_device_state_message *, device_state, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
@ -769,7 +692,11 @@ int ast_publish_device_state_full(
return -1;
}
message = stasis_message_create(ast_device_state_message_type(), device_state);
message = stasis_message_create_full(ast_device_state_message_type(), device_state,
eid);
if (!message) {
return -1;
}
device_specific_topic = ast_device_state_topic(device);
if (!device_specific_topic) {
@ -783,6 +710,7 @@ int ast_publish_device_state_full(
static const char *device_state_get_id(struct stasis_message *message)
{
struct ast_device_state_message *device_state;
if (ast_device_state_message_type() != stasis_message_type(message)) {
return NULL;
}
@ -792,20 +720,124 @@ static const char *device_state_get_id(struct stasis_message *message)
return NULL;
}
return device_state->cache_id;
return device_state->device;
}
/*!
* \internal
* \brief Callback to publish the aggregate device state cache entry message.
* \since 12.2.0
*
* \param cache_topic Caching topic the aggregate message may be published over.
* \param aggregate The aggregate shapshot message to publish.
*
* \return Nothing
*/
static void device_state_aggregate_publish(struct stasis_topic *cache_topic, struct stasis_message *aggregate)
{
const char *device;
struct stasis_topic *device_specific_topic;
device = device_state_get_id(aggregate);
if (!device) {
return;
}
device_specific_topic = ast_device_state_topic(device);
if (!device_specific_topic) {
return;
}
stasis_publish(device_specific_topic, aggregate);
}
/*!
* \internal
* \brief Callback to calculate the aggregate device state cache entry.
* \since 12.2.0
*
* \param entry Cache entry to calculate a new aggregate snapshot.
* \param new_snapshot The shapshot that is being updated.
*
* \note Return a ref bumped pointer from stasis_cache_entry_get_aggregate()
* if a new aggregate could not be calculated because of error.
*
* \return New aggregate-snapshot calculated on success.
* Caller has a reference on return.
*/
static struct stasis_message *device_state_aggregate_calc(struct stasis_cache_entry *entry, struct stasis_message *new_snapshot)
{
struct stasis_message *aggregate_snapshot;
struct stasis_message *snapshot;
struct ast_device_state_message *device_state;
const char *device = NULL;
struct ast_devstate_aggregate aggregate;
int idx;
/* Determine the new aggregate device state. */
ast_devstate_aggregate_init(&aggregate);
snapshot = stasis_cache_entry_get_local(entry);
if (snapshot) {
device_state = stasis_message_data(snapshot);
device = device_state->device;
ast_devstate_aggregate_add(&aggregate, device_state->state);
}
for (idx = 0; ; ++idx) {
snapshot = stasis_cache_entry_get_remote(entry, idx);
if (!snapshot) {
break;
}
device_state = stasis_message_data(snapshot);
device = device_state->device;
ast_devstate_aggregate_add(&aggregate, device_state->state);
}
if (!device) {
/* There are no device states cached. Delete the aggregate. */
return NULL;
}
snapshot = stasis_cache_entry_get_aggregate(entry);
if (snapshot) {
device_state = stasis_message_data(snapshot);
if (device_state->state == ast_devstate_aggregate_result(&aggregate)) {
/* Aggregate device state did not change. */
return ao2_bump(snapshot);
}
}
device_state = device_state_alloc(device, ast_devstate_aggregate_result(&aggregate),
AST_DEVSTATE_CACHABLE, NULL);
if (!device_state) {
/* Bummer. We have to keep the old aggregate snapshot. */
return ao2_bump(snapshot);
}
aggregate_snapshot = stasis_message_create_full(ast_device_state_message_type(),
device_state, NULL);
ao2_cleanup(device_state);
if (!aggregate_snapshot) {
/* Bummer. We have to keep the old aggregate snapshot. */
return ao2_bump(snapshot);
}
return aggregate_snapshot;
}
static void devstate_cleanup(void)
{
devstate_message_sub = stasis_unsubscribe_and_join(devstate_message_sub);
ao2_cleanup(device_state_topic_all);
device_state_topic_all = NULL;
device_state_topic_cached = stasis_caching_unsubscribe_and_join(device_state_topic_cached);
ao2_cleanup(device_state_cache);
device_state_cache = NULL;
device_state_topic_cached = stasis_caching_unsubscribe_and_join(device_state_topic_cached);
STASIS_MESSAGE_TYPE_CLEANUP(ast_device_state_message_type);
ao2_cleanup(device_state_topic_pool);
device_state_topic_pool = NULL;
ao2_cleanup(device_state_topic_all);
device_state_topic_all = NULL;
STASIS_MESSAGE_TYPE_CLEANUP(ast_device_state_message_type);
}
int devstate_init(void)
@ -817,25 +849,32 @@ int devstate_init(void)
}
device_state_topic_all = stasis_topic_create("ast_device_state_topic");
if (!device_state_topic_all) {
devstate_cleanup();
return -1;
}
device_state_cache = stasis_cache_create(device_state_get_id);
if (!device_state_cache) {
device_state_topic_pool = stasis_topic_pool_create(ast_device_state_topic_all());
if (!device_state_topic_pool) {
devstate_cleanup();
return -1;
}
device_state_topic_cached = stasis_caching_topic_create(device_state_topic_all, device_state_cache);
if (!device_state_topic_cached) {
device_state_cache = stasis_cache_create_full(device_state_get_id,
device_state_aggregate_calc, device_state_aggregate_publish);
if (!device_state_cache) {
devstate_cleanup();
return -1;
}
device_state_topic_pool = stasis_topic_pool_create(ast_device_state_topic_all());
if (!device_state_topic_pool) {
device_state_topic_cached = stasis_caching_topic_create(ast_device_state_topic_all(),
device_state_cache);
if (!device_state_topic_cached) {
devstate_cleanup();
return -1;
}
devstate_message_sub = stasis_subscribe(ast_device_state_topic_cached(), devstate_change_collector_cb, NULL);
devstate_message_sub = stasis_subscribe(ast_device_state_topic_all(),
devstate_change_cb, NULL);
if (!devstate_message_sub) {
ast_log(LOG_ERROR, "Failed to create subscription for the device state change collector\n");
ast_log(LOG_ERROR, "Failed to create subscription creating uncached device state aggregate events.\n");
devstate_cleanup();
return -1;
}

650
main/stasis_cache.c
Unescape View File

@ -36,6 +36,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
#include "asterisk/stasis_internal.h"
#include "asterisk/stasis.h"
#include "asterisk/utils.h"
#include "asterisk/vector.h"
#ifdef LOW_MEMORY
#define NUM_CACHE_BUCKETS 17
@ -47,6 +48,8 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
struct stasis_cache {
struct ao2_container *entries;
snapshot_get_id id_fn;
cache_aggregate_calc_fn aggregate_calc_fn;
cache_aggregate_publish_fn aggregate_publish_fn;
};
/*! \internal */
@ -124,29 +127,53 @@ struct stasis_caching_topic *stasis_caching_unsubscribe_and_join(struct stasis_c
return NULL;
}
struct cache_entry {
struct cache_entry_key {
struct stasis_message_type *type;
char *id;
struct stasis_message *snapshot;
const char *id;
};
struct stasis_cache_entry {
struct cache_entry_key key;
/*! Aggregate snapshot of the stasis cache. */
struct stasis_message *aggregate;
/*! Local entity snapshot of the stasis event. */
struct stasis_message *local;
/*! Remote entity snapshots of the stasis event. */
AST_VECTOR(, struct stasis_message *) remote;
};
static void cache_entry_dtor(void *obj)
{
struct cache_entry *entry = obj;
ao2_cleanup(entry->type);
entry->type = NULL;
ast_free(entry->id);
entry->id = NULL;
ao2_cleanup(entry->snapshot);
entry->snapshot = NULL;
struct stasis_cache_entry *entry = obj;
size_t idx;
ao2_cleanup(entry->key.type);
entry->key.type = NULL;
ast_free((char *) entry->key.id);
entry->key.id = NULL;
ao2_cleanup(entry->aggregate);
entry->aggregate = NULL;
ao2_cleanup(entry->local);
entry->local = NULL;
for (idx = 0; idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
struct stasis_message *remote;
remote = AST_VECTOR_GET(&entry->remote, idx);
ao2_cleanup(remote);
}
AST_VECTOR_FREE(&entry->remote);
}
static struct cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
static struct stasis_cache_entry *cache_entry_create(struct stasis_message_type *type, const char *id, struct stasis_message *snapshot)
{
RAII_VAR(struct cache_entry *, entry, NULL, ao2_cleanup);
struct stasis_cache_entry *entry;
int is_remote;
ast_assert(type != NULL);
ast_assert(id != NULL);
ast_assert(snapshot != NULL);
entry = ao2_alloc_options(sizeof(*entry), cache_entry_dtor,
AO2_ALLOC_OPT_LOCK_NOLOCK);
@ -154,173 +181,528 @@ static struct cache_entry *cache_entry_create(struct stasis_message_type *type,
return NULL;
}
entry->id = ast_strdup(id);
if (!entry->id) {
entry->key.id = ast_strdup(id);
if (!entry->key.id) {
ao2_cleanup(entry);
return NULL;
}
entry->key.type = ao2_bump(type);
is_remote = ast_eid_cmp(&ast_eid_default, stasis_message_eid(snapshot)) ? 1 : 0;
if (AST_VECTOR_INIT(&entry->remote, is_remote)) {
ao2_cleanup(entry);
return NULL;
}
ao2_ref(type, +1);
entry->type = type;
if (snapshot != NULL) {
ao2_ref(snapshot, +1);
entry->snapshot = snapshot;
if (is_remote) {
if (AST_VECTOR_APPEND(&entry->remote, snapshot)) {
ao2_cleanup(entry);
return NULL;
}
} else {
entry->local = snapshot;
}
ao2_bump(snapshot);
ao2_ref(entry, +1);
return entry;
}
static int cache_entry_hash(const void *obj, int flags)
{
const struct cache_entry *entry = obj;
const struct stasis_cache_entry *object;
const struct cache_entry_key *key;
int hash = 0;
ast_assert(!(flags & OBJ_KEY));
switch (flags & OBJ_SEARCH_MASK) {
case OBJ_SEARCH_KEY:
key = obj;
break;
case OBJ_SEARCH_OBJECT:
object = obj;
key = &object->key;
break;
default:
/* Hash can only work on something with a full key. */
ast_assert(0);
return 0;
}
hash += ast_hashtab_hash_string(stasis_message_type_name(entry->type));
hash += ast_hashtab_hash_string(entry->id);
hash += ast_hashtab_hash_string(stasis_message_type_name(key->type));
hash += ast_hashtab_hash_string(key->id);
return hash;
}
static int cache_entry_cmp(void *obj, void *arg, int flags)
{
const struct cache_entry *left = obj;
const struct cache_entry *right = arg;
const struct stasis_cache_entry *object_left = obj;
const struct stasis_cache_entry *object_right = arg;
const struct cache_entry_key *right_key = obj;
int cmp;
switch (flags & OBJ_SEARCH_MASK) {
case OBJ_SEARCH_OBJECT:
right_key = &object_right->key;
/* Fall through */
case OBJ_SEARCH_KEY:
cmp = object_left->key.type != right_key->type
|| strcmp(object_left->key.id, right_key->id);
break;
case OBJ_SEARCH_PARTIAL_KEY:
/* Not supported by container */
ast_assert(0);
cmp = -1;
break;
default:
/*
* What arg points to is specific to this traversal callback
* and has no special meaning to astobj2.
*/
cmp = 0;
break;
}
if (cmp) {
return 0;
}
/*
* At this point the traversal callback is identical to a sorted
* container.
*/
return CMP_MATCH;
}
static void cache_dtor(void *obj)
{
struct stasis_cache *cache = obj;
ast_assert(!(flags & OBJ_KEY));
ao2_cleanup(cache->entries);
cache->entries = NULL;
}
if (left->type == right->type && strcmp(left->id, right->id) == 0) {
return CMP_MATCH | CMP_STOP;
struct stasis_cache *stasis_cache_create_full(snapshot_get_id id_fn,
cache_aggregate_calc_fn aggregate_calc_fn,
cache_aggregate_publish_fn aggregate_publish_fn)
{
struct stasis_cache *cache;
cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
AO2_ALLOC_OPT_LOCK_NOLOCK);
if (!cache) {
return NULL;
}
return 0;
cache->entries = ao2_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
NUM_CACHE_BUCKETS, cache_entry_hash, NULL, cache_entry_cmp);
if (!cache->entries) {
ao2_cleanup(cache);
return NULL;
}
cache->id_fn = id_fn;
cache->aggregate_calc_fn = aggregate_calc_fn;
cache->aggregate_publish_fn = aggregate_publish_fn;
return cache;
}
static void cache_dtor(void *obj)
struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn)
{
struct stasis_cache *cache = obj;
return stasis_cache_create_full(id_fn, NULL, NULL);
}
ao2_cleanup(cache->entries);
cache->entries = NULL;
struct stasis_message *stasis_cache_entry_get_aggregate(struct stasis_cache_entry *entry)
{
return entry->aggregate;
}
struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn)
struct stasis_message *stasis_cache_entry_get_local(struct stasis_cache_entry *entry)
{
RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
return entry->local;
}
cache = ao2_alloc_options(sizeof(*cache), cache_dtor,
AO2_ALLOC_OPT_LOCK_NOLOCK);
if (!cache) {
return NULL;
}
struct stasis_message *stasis_cache_entry_get_remote(struct stasis_cache_entry *entry, int idx)
{
if (idx < AST_VECTOR_SIZE(&entry->remote)) {
return AST_VECTOR_GET(&entry->remote, idx);
}
return NULL;
}
cache->entries = ao2_container_alloc(NUM_CACHE_BUCKETS, cache_entry_hash,
cache_entry_cmp);
if (!cache->entries) {
return NULL;
}
/*!
* \internal
* \brief Find the cache entry in the cache entries container.
*
* \param entries Container of cached entries.
* \param type Type of message to retrieve the cache entry.
* \param id Identity of the snapshot to retrieve the cache entry.
*
* \note The entries container is already locked.
*
* \retval Cache-entry on success.
* \retval NULL Not in cache.
*/
static struct stasis_cache_entry *cache_find(struct ao2_container *entries, struct stasis_message_type *type, const char *id)
{
struct cache_entry_key search_key;
search_key.type = type;
search_key.id = id;
return ao2_find(entries, &search_key, OBJ_SEARCH_KEY | OBJ_NOLOCK);
}
/*!
* \internal
* \brief Remove the stasis snapshot in the cache entry determined by eid.
*
* \param entries Container of cached entries.
* \param cached_entry The entry to remove the snapshot from.
* \param eid Which snapshot in the cached entry.
*
* \note The entries container is already locked.
*
* \return Previous stasis entry snapshot.
*/
static struct stasis_message *cache_remove(struct ao2_container *entries, struct stasis_cache_entry *cached_entry, const struct ast_eid *eid)
{
struct stasis_message *old_snapshot;
int is_remote;
is_remote = ast_eid_cmp(eid, &ast_eid_default);
if (!is_remote) {
old_snapshot = cached_entry->local;
cached_entry->local = NULL;
} else {
int idx;
old_snapshot = NULL;
for (idx = 0; idx < AST_VECTOR_SIZE(&cached_entry->remote); ++idx) {
struct stasis_message *cur;
cur = AST_VECTOR_GET(&cached_entry->remote, idx);
if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
old_snapshot = AST_VECTOR_REMOVE_UNORDERED(&cached_entry->remote, idx);
break;
}
}
}
if (!cached_entry->local && !AST_VECTOR_SIZE(&cached_entry->remote)) {
ao2_unlink_flags(entries, cached_entry, OBJ_NOLOCK);
}
return old_snapshot;
}
cache->id_fn = id_fn;
/*!
* \internal
* \brief Update the stasis snapshot in the cache entry determined by eid.
*
* \param cached_entry The entry to remove the snapshot from.
* \param eid Which snapshot in the cached entry.
* \param new_snapshot Snapshot to replace the old snapshot.
*
* \return Previous stasis entry snapshot.
*/
static struct stasis_message *cache_udpate(struct stasis_cache_entry *cached_entry, const struct ast_eid *eid, struct stasis_message *new_snapshot)
{
struct stasis_message *old_snapshot;
int is_remote;
int idx;
is_remote = ast_eid_cmp(eid, &ast_eid_default);
if (!is_remote) {
old_snapshot = cached_entry->local;
cached_entry->local = ao2_bump(new_snapshot);
return old_snapshot;
}
old_snapshot = NULL;
for (idx = 0; idx < AST_VECTOR_SIZE(&cached_entry->remote); ++idx) {
struct stasis_message *cur;
ao2_ref(cache, +1);
return cache;
cur = AST_VECTOR_GET(&cached_entry->remote, idx);
if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
old_snapshot = AST_VECTOR_REMOVE_UNORDERED(&cached_entry->remote, idx);
break;
}
}
if (!AST_VECTOR_APPEND(&cached_entry->remote, new_snapshot)) {
ao2_bump(new_snapshot);
}
return old_snapshot;
}
static struct stasis_message *cache_put(struct stasis_cache *cache,
struct stasis_message_type *type, const char *id,
struct cache_put_snapshots {
/*! Old cache eid snapshot. */
struct stasis_message *old;
/*! Old cache aggregate snapshot. */
struct stasis_message *aggregate_old;
/*! New cache aggregate snapshot. */
struct stasis_message *aggregate_new;
};
static struct cache_put_snapshots cache_put(struct stasis_cache *cache,
struct stasis_message_type *type, const char *id, const struct ast_eid *eid,
struct stasis_message *new_snapshot)
{
RAII_VAR(struct cache_entry *, new_entry, NULL, ao2_cleanup);
RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
struct stasis_message *old_snapshot = NULL;
struct stasis_cache_entry *cached_entry;
struct cache_put_snapshots snapshots;
ast_assert(cache->entries != NULL);
ast_assert(eid != NULL);/* Aggregate snapshots not allowed to be put directly. */
ast_assert(new_snapshot == NULL ||
type == stasis_message_type(new_snapshot));
new_entry = cache_entry_create(type, id, new_snapshot);
memset(&snapshots, 0, sizeof(snapshots));
ao2_wrlock(cache->entries);
cached_entry = cache_find(cache->entries, type, id);
if (new_snapshot == NULL) {
/* Remove entry from cache */
cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_UNLINK);
/* Update the eid snapshot. */
if (!new_snapshot) {
/* Remove snapshot from cache */
if (cached_entry) {
old_snapshot = cached_entry->snapshot;
cached_entry->snapshot = NULL;
snapshots.old = cache_remove(cache->entries, cached_entry, eid);
}
} else if (cached_entry) {
/* Update snapshot in cache */
snapshots.old = cache_udpate(cached_entry, eid, new_snapshot);
} else {
/* Insert/update cache */
SCOPED_AO2LOCK(lock, cache->entries);
cached_entry = ao2_find(cache->entries, new_entry, OBJ_POINTER | OBJ_NOLOCK);
/* Insert into the cache */
cached_entry = cache_entry_create(type, id, new_snapshot);
if (cached_entry) {
/* Update cache. Because objects are moving, no need to update refcounts. */
old_snapshot = cached_entry->snapshot;
cached_entry->snapshot = new_entry->snapshot;
new_entry->snapshot = NULL;
} else {
/* Insert into the cache */
ao2_link_flags(cache->entries, new_entry, OBJ_NOLOCK);
ao2_link_flags(cache->entries, cached_entry, OBJ_NOLOCK);
}
}
/* Update the aggregate snapshot. */
if (cache->aggregate_calc_fn && cached_entry) {
snapshots.aggregate_new = cache->aggregate_calc_fn(cached_entry, new_snapshot);
snapshots.aggregate_old = cached_entry->aggregate;
cached_entry->aggregate = ao2_bump(snapshots.aggregate_new);
}
return old_snapshot;
ao2_unlock(cache->entries);
ao2_cleanup(cached_entry);
return snapshots;
}
struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
/*!
* \internal
* \brief Dump all entity snapshots in the cache entry into the given container.
*
* \param snapshots Container to put all snapshots in the cache entry.
* \param entry Cache entry to use.
*
* \retval 0 on success.
* \retval non-zero on error.
*/
static int cache_entry_dump(struct ao2_container *snapshots, const struct stasis_cache_entry *entry)
{
int idx;
int err = 0;
ast_assert(snapshots != NULL);
ast_assert(entry != NULL);
/* The aggregate snapshot is not a snapshot from an entity. */
if (entry->local) {
err |= !ao2_link(snapshots, entry->local);
}
for (idx = 0; !err && idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
struct stasis_message *snapshot;
snapshot = AST_VECTOR_GET(&entry->remote, idx);
err |= !ao2_link(snapshots, snapshot);
}
return err;
}
struct ao2_container *stasis_cache_get_all(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
{
RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup);
RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
struct stasis_cache_entry *cached_entry;
struct ao2_container *found;
ast_assert(cache != NULL);
ast_assert(cache->entries != NULL);
ast_assert(type != NULL);
ast_assert(id != NULL);
search_entry = cache_entry_create(type, id, NULL);
if (search_entry == NULL) {
found = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
if (!found) {
return NULL;
}
cached_entry = ao2_find(cache->entries, search_entry, OBJ_POINTER);
if (cached_entry == NULL) {
return NULL;
ao2_rdlock(cache->entries);
cached_entry = cache_find(cache->entries, type, id);
if (cached_entry && cache_entry_dump(found, cached_entry)) {
ao2_cleanup(found);
found = NULL;
}
ao2_unlock(cache->entries);
ao2_cleanup(cached_entry);
return found;
}
/*!
* \internal
* \brief Retrieve an item from the cache entry for a specific eid.
*
* \param entry Cache entry to use.
* \param eid Specific entity id to retrieve. NULL for aggregate.
*
* \note The returned snapshot has not had its reference bumped.
*
* \retval Snapshot from the cache.
* \retval \c NULL if snapshot is not found.
*/
static struct stasis_message *cache_entry_by_eid(const struct stasis_cache_entry *entry, const struct ast_eid *eid)
{
int is_remote;
int idx;
if (!eid) {
/* Get aggregate. */
return entry->aggregate;
}
/* Get snapshot with specific eid. */
is_remote = ast_eid_cmp(eid, &ast_eid_default);
if (!is_remote) {
return entry->local;
}
ast_assert(cached_entry->snapshot != NULL);
ao2_ref(cached_entry->snapshot, +1);
return cached_entry->snapshot;
for (idx = 0; idx < AST_VECTOR_SIZE(&entry->remote); ++idx) {
struct stasis_message *cur;
cur = AST_VECTOR_GET(&entry->remote, idx);
if (!ast_eid_cmp(eid, stasis_message_eid(cur))) {
return cur;
}
}
return NULL;
}
struct stasis_message *stasis_cache_get_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const char *id, const struct ast_eid *eid)
{
struct stasis_cache_entry *cached_entry;
struct stasis_message *snapshot = NULL;
ast_assert(cache != NULL);
ast_assert(cache->entries != NULL);
ast_assert(type != NULL);
ast_assert(id != NULL);
ao2_rdlock(cache->entries);
cached_entry = cache_find(cache->entries, type, id);
if (cached_entry) {
snapshot = cache_entry_by_eid(cached_entry, eid);
ao2_bump(snapshot);
}
ao2_unlock(cache->entries);
ao2_cleanup(cached_entry);
return snapshot;
}
struct stasis_message *stasis_cache_get(struct stasis_cache *cache, struct stasis_message_type *type, const char *id)
{
return stasis_cache_get_by_eid(cache, type, id, &ast_eid_default);
}
struct cache_dump_data {
struct ao2_container *cached;
struct ao2_container *container;
struct stasis_message_type *type;
const struct ast_eid *eid;
};
static int cache_dump_cb(void *obj, void *arg, int flags)
static int cache_dump_by_eid_cb(void *obj, void *arg, int flags)
{
struct cache_dump_data *cache_dump = arg;
struct cache_entry *entry = obj;
struct stasis_cache_entry *entry = obj;
if (!cache_dump->type || entry->key.type == cache_dump->type) {
struct stasis_message *snapshot;
if (!cache_dump->type || entry->type == cache_dump->type) {
ao2_link(cache_dump->cached, entry->snapshot);
snapshot = cache_entry_by_eid(entry, cache_dump->eid);
if (snapshot) {
if (!ao2_link(cache_dump->container, snapshot)) {
ao2_cleanup(cache_dump->container);
cache_dump->container = NULL;
return CMP_STOP;
}
}
}
return 0;
}
struct ao2_container *stasis_cache_dump_by_eid(struct stasis_cache *cache, struct stasis_message_type *type, const struct ast_eid *eid)
{
struct cache_dump_data cache_dump;
ast_assert(cache != NULL);
ast_assert(cache->entries != NULL);
cache_dump.eid = eid;
cache_dump.type = type;
cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
if (!cache_dump.container) {
return NULL;
}
ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_by_eid_cb, &cache_dump);
return cache_dump.container;
}
struct ao2_container *stasis_cache_dump(struct stasis_cache *cache, struct stasis_message_type *type)
{
return stasis_cache_dump_by_eid(cache, type, &ast_eid_default);
}
static int cache_dump_all_cb(void *obj, void *arg, int flags)
{
struct cache_dump_data *cache_dump = arg;
struct stasis_cache_entry *entry = obj;
if (!cache_dump->type || entry->key.type == cache_dump->type) {
if (cache_entry_dump(cache_dump->container, entry)) {
ao2_cleanup(cache_dump->container);
cache_dump->container = NULL;
return CMP_STOP;
}
}
return 0;
}
struct ao2_container *stasis_cache_dump_all(struct stasis_cache *cache, struct stasis_message_type *type)
{
struct cache_dump_data cache_dump;
ast_assert(cache != NULL);
ast_assert(cache->entries != NULL);
cache_dump.eid = NULL;
cache_dump.type = type;
cache_dump.cached = ao2_container_alloc_options(
AO2_ALLOC_OPT_LOCK_NOLOCK, 1, NULL, NULL);
if (!cache_dump.cached) {
cache_dump.container = ao2_container_alloc_list(AO2_ALLOC_OPT_LOCK_NOLOCK, 0, NULL, NULL);
if (!cache_dump.container) {
return NULL;
}
ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_cb, &cache_dump);
return cache_dump.cached;
ao2_callback(cache->entries, OBJ_MULTIPLE | OBJ_NODATA, cache_dump_all_cb, &cache_dump);
return cache_dump.container;
}
STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type);
@ -380,9 +762,13 @@ static struct stasis_message *update_create(struct stasis_message *old_snapshot,
static void caching_topic_exec(void *data, struct stasis_subscription *sub,
struct stasis_message *message)
{
RAII_VAR(struct stasis_caching_topic *, caching_topic_needs_unref, NULL, ao2_cleanup);
struct stasis_caching_topic *caching_topic_needs_unref;
struct stasis_caching_topic *caching_topic = data;
const char *id = NULL;
struct stasis_message *msg;
struct stasis_message *msg_put;
struct stasis_message_type *msg_type;
const struct ast_eid *msg_eid;
const char *msg_id;
ast_assert(caching_topic != NULL);
ast_assert(caching_topic->topic != NULL);
@ -391,50 +777,62 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub,
if (stasis_subscription_final_message(sub, message)) {
caching_topic_needs_unref = caching_topic;
} else {
caching_topic_needs_unref = NULL;
}
/* Handle cache clear event */
if (stasis_cache_clear_type() == stasis_message_type(message)) {
RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
struct stasis_message *clear_msg = stasis_message_data(message);
const char *clear_id = caching_topic->cache->id_fn(clear_msg);
struct stasis_message_type *clear_type = stasis_message_type(clear_msg);
msg_type = stasis_message_type(message);
if (stasis_cache_clear_type() == msg_type) {
/* Cache clear event. */
msg_put = NULL;
msg = stasis_message_data(message);
msg_type = stasis_message_type(msg);
} else {
/* Normal cache update event. */
msg_put = message;
msg = message;
}
ast_assert(msg_type != NULL);
ast_assert(clear_type != NULL);
msg_eid = stasis_message_eid(msg);/* msg_eid is NULL for aggregate message. */
msg_id = caching_topic->cache->id_fn(msg);
if (msg_id && msg_eid) {
struct stasis_message *update;
struct cache_put_snapshots snapshots;
if (clear_id) {
old_snapshot = cache_put(caching_topic->cache, clear_type, clear_id, NULL);
if (old_snapshot) {
update = update_create(old_snapshot, NULL);
/* Update the cache */
snapshots = cache_put(caching_topic->cache, msg_type, msg_id, msg_eid, msg_put);
if (snapshots.old || msg_put) {
update = update_create(snapshots.old, msg_put);
if (update) {
stasis_publish(caching_topic->topic, update);
return;
}
ao2_cleanup(update);
} else {
ast_log(LOG_ERROR,
"Attempting to remove an item from the %s cache that isn't there: %s %s\n",
stasis_topic_name(caching_topic->topic), stasis_message_type_name(clear_type), clear_id);
return;
stasis_topic_name(caching_topic->topic),
stasis_message_type_name(msg_type), msg_id);
}
}
id = caching_topic->cache->id_fn(message);
if (id == NULL) {
/* Object isn't cached; discard */
} else {
/* Update the cache */
RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, update, NULL, ao2_cleanup);
old_snapshot = cache_put(caching_topic->cache, stasis_message_type(message), id, message);
update = update_create(old_snapshot, message);
if (update == NULL) {
return;
if (snapshots.aggregate_old != snapshots.aggregate_new) {
if (snapshots.aggregate_new && caching_topic->cache->aggregate_publish_fn) {
caching_topic->cache->aggregate_publish_fn(caching_topic->original_topic,
snapshots.aggregate_new);
}
update = update_create(snapshots.aggregate_old, snapshots.aggregate_new);
if (update) {
stasis_publish(caching_topic->topic, update);
}
ao2_cleanup(update);
}
stasis_publish(caching_topic->topic, update);
ao2_cleanup(snapshots.old);
ao2_cleanup(snapshots.aggregate_old);
ao2_cleanup(snapshots.aggregate_new);
}
ao2_cleanup(caching_topic_needs_unref);
}
struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, struct stasis_cache *cache)

30
main/stasis_message.c
Unescape View File

@ -53,7 +53,7 @@ static void message_type_dtor(void *obj)
struct stasis_message_type *stasis_message_type_create(const char *name,
struct stasis_message_vtable *vtable)
{
RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
struct stasis_message_type *type;
type = ao2_alloc(sizeof(*type), message_type_dtor);
if (!type) {
@ -66,11 +66,11 @@ struct stasis_message_type *stasis_message_type_create(const char *name,
type->name = ast_strdup(name);
if (!type->name) {
ao2_cleanup(type);
return NULL;
}
type->vtable = vtable;
ao2_ref(type, +1);
return type;
}
@ -85,8 +85,12 @@ struct stasis_message {
struct timeval timestamp;
/*! Type of the message */
struct stasis_message_type *type;
/*! Where this message originated. NULL if aggregate message. */
const struct ast_eid *eid_ptr;
/*! Message content */
void *data;
/*! Where this message originated. */
struct ast_eid eid;
};
static void stasis_message_dtor(void *obj)
@ -96,9 +100,9 @@ static void stasis_message_dtor(void *obj)
ao2_cleanup(message->data);
}
struct stasis_message *stasis_message_create(struct stasis_message_type *type, void *data)
struct stasis_message *stasis_message_create_full(struct stasis_message_type *type, void *data, const struct ast_eid *eid)
{
RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
struct stasis_message *message;
if (type == NULL || data == NULL) {
return NULL;
@ -114,11 +118,27 @@ struct stasis_message *stasis_message_create(struct stasis_message_type *type, v
message->type = type;
ao2_ref(data, +1);
message->data = data;
if (eid) {
message->eid_ptr = &message->eid;
message->eid = *eid;
}
ao2_ref(message, +1);
return message;
}
struct stasis_message *stasis_message_create(struct stasis_message_type *type, void *data)
{
return stasis_message_create_full(type, data, &ast_eid_default);
}
const struct ast_eid *stasis_message_eid(const struct stasis_message *msg)
{
if (msg == NULL) {
return NULL;
}
return msg->eid_ptr;
}
struct stasis_message_type *stasis_message_type(const struct stasis_message *msg)
{
if (msg == NULL) {

141
tests/test_devicestate.c
Unescape View File

@ -277,45 +277,49 @@ AST_TEST_DEFINE(device2extenstate_test)
}
struct consumer {
ast_mutex_t lock;
ast_cond_t out;
int already_out;
int sig_on_non_aggregate_state;
int event_count;
enum ast_device_state state;
enum ast_device_state aggregate_state;
int sig_on_non_aggregate_state;
};
static void consumer_dtor(void *obj) {
static void consumer_dtor(void *obj)
{
struct consumer *consumer = obj;
ast_mutex_destroy(&consumer->lock);
ast_cond_destroy(&consumer->out);
}
static struct consumer *consumer_create(void) {
RAII_VAR(struct consumer *, consumer, NULL, ao2_cleanup);
static void consumer_reset(struct consumer *consumer)
{
consumer->already_out = 0;
consumer->event_count = 0;
consumer->state = AST_DEVICE_TOTAL;
consumer->aggregate_state = AST_DEVICE_TOTAL;
}
consumer = ao2_alloc(sizeof(*consumer), consumer_dtor);
static struct consumer *consumer_create(void)
{
struct consumer *consumer;
consumer = ao2_alloc(sizeof(*consumer), consumer_dtor);
if (!consumer) {
return NULL;
}
ast_mutex_init(&consumer->lock);
ast_cond_init(&consumer->out, NULL);
consumer->sig_on_non_aggregate_state = 0;
consumer_reset(consumer);
ao2_ref(consumer, +1);
return consumer;
}
static void consumer_exec(void *data, struct stasis_subscription *sub, struct stasis_message *message)
{
struct consumer *consumer = data;
RAII_VAR(struct consumer *, consumer_needs_cleanup, NULL, ao2_cleanup);
struct stasis_cache_update *cache_update = stasis_message_data(message);
struct ast_device_state_message *device_state;
SCOPED_MUTEX(lock, &consumer->lock);
if (!cache_update->new_snapshot) {
return;
@ -328,17 +332,22 @@ static void consumer_exec(void *data, struct stasis_subscription *sub, struct st
return;
}
if (device_state->eid) {
consumer->state = device_state->state;
if (consumer->sig_on_non_aggregate_state) {
consumer->sig_on_non_aggregate_state = 0;
{
SCOPED_AO2LOCK(lock, consumer);
++consumer->event_count;
if (device_state->eid) {
consumer->state = device_state->state;
if (consumer->sig_on_non_aggregate_state) {
consumer->sig_on_non_aggregate_state = 0;
consumer->already_out = 1;
ast_cond_signal(&consumer->out);
}
} else {
consumer->aggregate_state = device_state->state;
consumer->already_out = 1;
ast_cond_signal(&consumer->out);
}
} else {
consumer->aggregate_state = device_state->state;
consumer->already_out = 1;
ast_cond_signal(&consumer->out);
}
}
@ -360,45 +369,46 @@ static void consumer_wait_for(struct consumer *consumer)
.tv_nsec = start.tv_usec * 1000
};
SCOPED_MUTEX(lock, &consumer->lock);
SCOPED_AO2LOCK(lock, consumer);
if (consumer->already_out) {
consumer->already_out = 0;
}
while(1) {
res = ast_cond_timedwait(&consumer->out, &consumer->lock, &end);
while (!consumer->already_out) {
res = ast_cond_timedwait(&consumer->out, ao2_object_get_lockaddr(consumer), &end);
if (!res || res == ETIMEDOUT) {
break;
}
}
consumer->already_out = 0;
}
static int remove_device_states_cb(void *obj, void *arg, int flags)
{
RAII_VAR(struct stasis_message *, msg, obj, ao2_cleanup);
struct stasis_message *msg = obj;
struct ast_device_state_message *device_state = stasis_message_data(msg);
if (strcmp(UNIT_TEST_DEVICE_IDENTIFIER, device_state->device)) {
msg = NULL;
/* Not a unit test device */
return 0;
}
msg = stasis_cache_clear_create(msg);
/* topic guaranteed to have been created by this point */
stasis_publish(ast_device_state_topic(device_state->device), msg);
if (msg) {
/* topic guaranteed to have been created by this point */
stasis_publish(ast_device_state_topic(device_state->device), msg);
}
ao2_cleanup(msg);
return 0;
}
static void cache_cleanup(int unused)
{
RAII_VAR(struct ao2_container *, cache_dump, NULL, ao2_cleanup);
struct ao2_container *cache_dump;
/* remove all device states created during this test */
cache_dump = stasis_cache_dump(ast_device_state_cache(), NULL);
cache_dump = stasis_cache_dump_all(ast_device_state_cache(), NULL);
if (!cache_dump) {
return;
}
ao2_callback(cache_dump, 0, remove_device_states_cb, NULL);
ao2_cleanup(cache_dump);
}
AST_TEST_DEFINE(device_state_aggregation_test)
@ -407,9 +417,9 @@ AST_TEST_DEFINE(device_state_aggregation_test)
RAII_VAR(struct stasis_message_router *, device_msg_router, NULL, stasis_message_router_unsubscribe);
RAII_VAR(struct ast_eid *, foreign_eid, NULL, ast_free);
RAII_VAR(int, cleanup_cache, 0, cache_cleanup);
RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
int res;
struct ast_device_state_message *device_state;
struct stasis_message *msg;
switch (cmd) {
case TEST_INIT:
@ -447,56 +457,67 @@ AST_TEST_DEFINE(device_state_aggregation_test)
/* push local state */
ast_publish_device_state(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_NOT_INUSE, AST_DEVSTATE_CACHABLE);
/* Check cache aggregate state immediately */
ao2_cleanup(msg);
msg = stasis_cache_get_by_eid(ast_device_state_cache(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER, NULL);
device_state = stasis_message_data(msg);
ast_test_validate(test, AST_DEVICE_NOT_INUSE == device_state->state);
consumer_wait_for(consumer);
ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->state);
ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->aggregate_state);
msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER);
device_state = stasis_message_data(msg);
ast_test_validate(test, AST_DEVICE_NOT_INUSE == device_state->state);
ao2_cleanup(msg);
msg = NULL;
ast_test_validate(test, 2 == consumer->event_count);
consumer_reset(consumer);
/* push remote state */
/* this will not produce a new aggregate state message since the aggregate state does not change */
consumer->sig_on_non_aggregate_state = 1;
ast_publish_device_state_full(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_NOT_INUSE, AST_DEVSTATE_CACHABLE, foreign_eid);
consumer_wait_for(consumer);
ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->state);
ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->aggregate_state);
msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER);
/* Check cache aggregate state immediately */
ao2_cleanup(msg);
msg = stasis_cache_get_by_eid(ast_device_state_cache(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER, NULL);
device_state = stasis_message_data(msg);
ast_test_validate(test, AST_DEVICE_NOT_INUSE == device_state->state);
ao2_cleanup(msg);
msg = NULL;
/* Check for expected events. */
consumer_wait_for(consumer);
ast_test_validate(test, AST_DEVICE_NOT_INUSE == consumer->state);
ast_test_validate(test, AST_DEVICE_TOTAL == consumer->aggregate_state);
ast_test_validate(test, 1 == consumer->event_count);
consumer_reset(consumer);
/* push remote state different from local state */
ast_publish_device_state_full(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_INUSE, AST_DEVSTATE_CACHABLE, foreign_eid);
/* Check cache aggregate state immediately */
ao2_cleanup(msg);
msg = stasis_cache_get_by_eid(ast_device_state_cache(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER, NULL);
device_state = stasis_message_data(msg);
ast_test_validate(test, AST_DEVICE_INUSE == device_state->state);
/* Check for expected events. */
consumer_wait_for(consumer);
ast_test_validate(test, AST_DEVICE_INUSE == consumer->state);
ast_test_validate(test, AST_DEVICE_INUSE == consumer->aggregate_state);
msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER);
device_state = stasis_message_data(msg);
ast_test_validate(test, AST_DEVICE_INUSE == device_state->state);
ao2_cleanup(msg);
msg = NULL;
ast_test_validate(test, 2 == consumer->event_count);
consumer_reset(consumer);
/* push local state that will cause aggregated state different from local non-aggregate state */
ast_publish_device_state(UNIT_TEST_DEVICE_IDENTIFIER, AST_DEVICE_RINGING, AST_DEVSTATE_CACHABLE);
/* Check cache aggregate state immediately */
ao2_cleanup(msg);
msg = stasis_cache_get_by_eid(ast_device_state_cache(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER, NULL);
device_state = stasis_message_data(msg);
ast_test_validate(test, AST_DEVICE_RINGINUSE == device_state->state);
/* Check for expected events. */
consumer_wait_for(consumer);
ast_test_validate(test, AST_DEVICE_RINGING == consumer->state);
ast_test_validate(test, AST_DEVICE_RINGINUSE == consumer->aggregate_state);
msg = stasis_cache_get(ast_device_state_cache(), ast_device_state_message_type(), UNIT_TEST_DEVICE_IDENTIFIER);
device_state = stasis_message_data(msg);
ast_test_validate(test, AST_DEVICE_RINGINUSE == device_state->state);
ao2_cleanup(msg);
msg = NULL;
ast_test_validate(test, 2 == consumer->event_count);
consumer_reset(consumer);
return AST_TEST_PASS;
}

396
tests/test_stasis.c
Unescape View File

@ -94,11 +94,13 @@ AST_TEST_DEFINE(message_type)
AST_TEST_DEFINE(message)
{
RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, uut, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, uut1, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, uut2, NULL, ao2_cleanup);
RAII_VAR(char *, data, NULL, ao2_cleanup);
char *expected = "SomeData";
struct timeval expected_timestamp;
struct timeval time_diff;
struct ast_eid foreign_eid;
switch (cmd) {
case TEST_INIT:
@ -112,29 +114,42 @@ AST_TEST_DEFINE(message)
}
memset(&foreign_eid, 0xFF, sizeof(foreign_eid));
type = stasis_message_type_create("SomeMessage", NULL);
ast_test_validate(test, NULL == stasis_message_create(NULL, NULL));
ast_test_validate(test, NULL == stasis_message_create(type, NULL));
ast_test_validate(test, NULL == stasis_message_create_full(NULL, NULL, NULL));
ast_test_validate(test, NULL == stasis_message_create_full(type, NULL, NULL));
data = ao2_alloc(strlen(expected) + 1, NULL);
strcpy(data, expected);
strcpy(data, expected);/* Safe */
expected_timestamp = ast_tvnow();
uut = stasis_message_create(type, data);
ast_test_validate(test, NULL != uut);
ast_test_validate(test, type == stasis_message_type(uut));
ast_test_validate(test, 0 == strcmp(expected, stasis_message_data(uut)));
ast_test_validate(test, 2 == ao2_ref(data, 0)); /* uut has ref to data */
time_diff = ast_tvsub(*stasis_message_timestamp(uut), expected_timestamp);
uut1 = stasis_message_create_full(type, data, &foreign_eid);
uut2 = stasis_message_create_full(type, data, NULL);
ast_test_validate(test, NULL != uut1);
ast_test_validate(test, NULL != uut2);
ast_test_validate(test, type == stasis_message_type(uut1));
ast_test_validate(test, type == stasis_message_type(uut2));
ast_test_validate(test, 0 == strcmp(expected, stasis_message_data(uut1)));
ast_test_validate(test, 0 == strcmp(expected, stasis_message_data(uut2)));
ast_test_validate(test, NULL != stasis_message_eid(uut1));
ast_test_validate(test, NULL == stasis_message_eid(uut2));
ast_test_validate(test, !ast_eid_cmp(&foreign_eid, stasis_message_eid(uut1)));
ast_test_validate(test, 3 == ao2_ref(data, 0)); /* uut1 and uut2 have ref to data */
time_diff = ast_tvsub(*stasis_message_timestamp(uut1), expected_timestamp);
/* 10ms is certainly long enough for the two calls to complete */
ast_test_validate(test, time_diff.tv_sec == 0);
ast_test_validate(test, time_diff.tv_usec < 10000);
ao2_ref(uut, -1);
uut = NULL;
ast_test_validate(test, 1 == ao2_ref(data, 0)); /* uut unreffed data */
ao2_ref(uut1, -1);
uut1 = NULL;
ast_test_validate(test, 2 == ao2_ref(data, 0)); /* uut1 unreffed data */
ao2_ref(uut2, -1);
uut2 = NULL;
ast_test_validate(test, 1 == ao2_ref(data, 0)); /* uut2 unreffed data */
return AST_TEST_PASS;
}
@ -643,11 +658,12 @@ struct cache_test_data {
static void cache_test_data_dtor(void *obj)
{
struct cache_test_data *data = obj;
ast_free(data->id);
ast_free(data->value);
}
static struct stasis_message *cache_test_message_create(struct stasis_message_type *type, const char *name, const char *value)
static struct stasis_message *cache_test_message_create_full(struct stasis_message_type *type, const char *name, const char *value, struct ast_eid *eid)
{
RAII_VAR(struct cache_test_data *, data, NULL, ao2_cleanup);
@ -665,7 +681,12 @@ static struct stasis_message *cache_test_message_create(struct stasis_message_ty
return NULL;
}
return stasis_message_create(type, data);
return stasis_message_create_full(type, data, eid);
}
static struct stasis_message *cache_test_message_create(struct stasis_message_type *type, const char *name, const char *value)
{
return cache_test_message_create_full(type, name, value, &ast_eid_default);
}
static const char *cache_test_data_id(struct stasis_message *message)
@ -678,6 +699,81 @@ static const char *cache_test_data_id(struct stasis_message *message)
return cachable->id;
}
static struct stasis_message *cache_test_aggregate_calc_fn(struct stasis_cache_entry *entry, struct stasis_message *new_snapshot)
{
struct stasis_message *aggregate_snapshot;
struct stasis_message *snapshot;
struct stasis_message_type *type = NULL;
struct cache_test_data *test_data = NULL;
int idx;
int accumulated = 0;
char aggregate_str[30];
/* Accumulate the aggregate value. */
snapshot = stasis_cache_entry_get_local(entry);
if (snapshot) {
type = stasis_message_type(snapshot);
test_data = stasis_message_data(snapshot);
accumulated += atoi(test_data->value);
}
for (idx = 0; ; ++idx) {
snapshot = stasis_cache_entry_get_remote(entry, idx);
if (!snapshot) {
break;
}
type = stasis_message_type(snapshot);
test_data = stasis_message_data(snapshot);
accumulated += atoi(test_data->value);
}
if (!test_data) {
/* There are no test entries cached. Delete the aggregate. */
return NULL;
}
snapshot = stasis_cache_entry_get_aggregate(entry);
if (snapshot) {
type = stasis_message_type(snapshot);
test_data = stasis_message_data(snapshot);
if (accumulated == atoi(test_data->value)) {
/* Aggregate test entry did not change. */
return ao2_bump(snapshot);
}
}
snprintf(aggregate_str, sizeof(aggregate_str), "%d", accumulated);
aggregate_snapshot = cache_test_message_create_full(type, test_data->id, aggregate_str, NULL);
if (!aggregate_snapshot) {
/* Bummer. We have to keep the old aggregate snapshot. */
ast_log(LOG_ERROR, "Could not create aggregate snapshot.\n");
return ao2_bump(snapshot);
}
return aggregate_snapshot;
}
static void cache_test_aggregate_publish_fn(struct stasis_topic *topic, struct stasis_message *aggregate)
{
stasis_publish(topic, aggregate);
}
static int check_cache_aggregate(struct stasis_cache *cache, struct stasis_message_type *cache_type, const char *id, const char *value)
{
RAII_VAR(struct stasis_message *, aggregate, NULL, ao2_cleanup);
struct cache_test_data *test_data;
aggregate = stasis_cache_get_by_eid(cache, cache_type, id, NULL);
if (!aggregate) {
/* No aggregate, return true if given no value. */
return !value;
}
/* Return true if the given value matches the aggregate value. */
test_data = stasis_message_data(aggregate);
return value && !strcmp(value, test_data->value);
}
AST_TEST_DEFINE(cache_filter)
{
RAII_VAR(struct stasis_message_type *, non_cache_type, NULL, ao2_cleanup);
@ -845,8 +941,8 @@ AST_TEST_DEFINE(cache_dump)
case TEST_INIT:
info->name = __func__;
info->category = test_category;
info->summary = "Test passing messages through cache topic unscathed.";
info->description = "Test passing messages through cache topic unscathed.";
info->summary = "Test cache dump routines.";
info->description = "Test cache dump routines.";
return AST_TEST_NOT_RUN;
case TEST_EXECUTE:
break;
@ -937,6 +1033,266 @@ AST_TEST_DEFINE(cache_dump)
return AST_TEST_PASS;
}
AST_TEST_DEFINE(cache_eid_aggregate)
{
RAII_VAR(struct stasis_message_type *, cache_type, NULL, ao2_cleanup);
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
RAII_VAR(struct stasis_cache *, cache, NULL, ao2_cleanup);
RAII_VAR(struct stasis_caching_topic *, caching_topic, NULL, stasis_caching_unsubscribe);
RAII_VAR(struct consumer *, cache_consumer, NULL, ao2_cleanup);
RAII_VAR(struct consumer *, topic_consumer, NULL, ao2_cleanup);
RAII_VAR(struct stasis_subscription *, topic_sub, NULL, stasis_unsubscribe);
RAII_VAR(struct stasis_subscription *, cache_sub, NULL, stasis_unsubscribe);
RAII_VAR(struct stasis_message *, test_message1_1, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, test_message2_1, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, test_message2_2, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, test_message2_3, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, test_message2_4, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, test_message1_clear, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, test_message2_clear, NULL, ao2_cleanup);
RAII_VAR(struct ao2_container *, cache_dump, NULL, ao2_cleanup);
int actual_len;
struct ao2_iterator i;
void *obj;
struct ast_eid foreign_eid1;
struct ast_eid foreign_eid2;
switch (cmd) {
case TEST_INIT:
info->name = __func__;
info->category = test_category;
info->summary = "Test cache eid and aggregate support.";
info->description = "Test cache eid and aggregate support.";
return AST_TEST_NOT_RUN;
case TEST_EXECUTE:
break;
}
memset(&foreign_eid1, 0xAA, sizeof(foreign_eid1));
memset(&foreign_eid2, 0xBB, sizeof(foreign_eid2));
cache_type = stasis_message_type_create("Cacheable", NULL);
ast_test_validate(test, NULL != cache_type);
topic = stasis_topic_create("SomeTopic");
ast_test_validate(test, NULL != topic);
/* To consume events published to the topic. */
topic_consumer = consumer_create(1);
ast_test_validate(test, NULL != topic_consumer);
topic_sub = stasis_subscribe(topic, consumer_exec, topic_consumer);
ast_test_validate(test, NULL != topic_sub);
ao2_ref(topic_consumer, +1);
cache = stasis_cache_create_full(cache_test_data_id,
cache_test_aggregate_calc_fn, cache_test_aggregate_publish_fn);
ast_test_validate(test, NULL != cache);
caching_topic = stasis_caching_topic_create(topic, cache);
ast_test_validate(test, NULL != caching_topic);
/* To consume update events published to the caching_topic. */
cache_consumer = consumer_create(1);
ast_test_validate(test, NULL != cache_consumer);
cache_sub = stasis_subscribe(stasis_caching_get_topic(caching_topic), consumer_exec, cache_consumer);
ast_test_validate(test, NULL != cache_sub);
ao2_ref(cache_consumer, +1);
/* Create test messages. */
test_message1_1 = cache_test_message_create_full(cache_type, "1", "1", &ast_eid_default);
ast_test_validate(test, NULL != test_message1_1);
test_message2_1 = cache_test_message_create_full(cache_type, "2", "1", &ast_eid_default);
ast_test_validate(test, NULL != test_message2_1);
test_message2_2 = cache_test_message_create_full(cache_type, "2", "2", &foreign_eid1);
ast_test_validate(test, NULL != test_message2_2);
test_message2_3 = cache_test_message_create_full(cache_type, "2", "3", &foreign_eid2);
ast_test_validate(test, NULL != test_message2_3);
test_message2_4 = cache_test_message_create_full(cache_type, "2", "4", &foreign_eid2);
ast_test_validate(test, NULL != test_message2_4);
/* Post some snapshots */
stasis_publish(topic, test_message1_1);
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "1", "1"));
stasis_publish(topic, test_message2_1);
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "1"));
stasis_publish(topic, test_message2_2);
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "3"));
actual_len = consumer_wait_for(cache_consumer, 6);
ast_test_validate(test, 6 == actual_len);
actual_len = consumer_wait_for(topic_consumer, 6);
ast_test_validate(test, 6 == actual_len);
/* Check the cache */
ao2_cleanup(cache_dump);
cache_dump = stasis_cache_dump_all(cache, NULL);
ast_test_validate(test, NULL != cache_dump);
ast_test_validate(test, 3 == ao2_container_count(cache_dump));
i = ao2_iterator_init(cache_dump, 0);
while ((obj = ao2_iterator_next(&i))) {
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
ast_test_validate(test,
actual_cache_entry == test_message1_1
|| actual_cache_entry == test_message2_1
|| actual_cache_entry == test_message2_2);
}
ao2_iterator_destroy(&i);
/* Check the local cached items */
ao2_cleanup(cache_dump);
cache_dump = stasis_cache_dump_by_eid(cache, NULL, &ast_eid_default);
ast_test_validate(test, NULL != cache_dump);
ast_test_validate(test, 2 == ao2_container_count(cache_dump));
i = ao2_iterator_init(cache_dump, 0);
while ((obj = ao2_iterator_next(&i))) {
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
ast_test_validate(test,
actual_cache_entry == test_message1_1
|| actual_cache_entry == test_message2_1);
}
ao2_iterator_destroy(&i);
/* Post snapshot 2 from another eid. */
stasis_publish(topic, test_message2_3);
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "6"));
actual_len = consumer_wait_for(cache_consumer, 8);
ast_test_validate(test, 8 == actual_len);
actual_len = consumer_wait_for(topic_consumer, 8);
ast_test_validate(test, 8 == actual_len);
/* Check the cache */
ao2_cleanup(cache_dump);
cache_dump = stasis_cache_dump_all(cache, NULL);
ast_test_validate(test, NULL != cache_dump);
ast_test_validate(test, 4 == ao2_container_count(cache_dump));
i = ao2_iterator_init(cache_dump, 0);
while ((obj = ao2_iterator_next(&i))) {
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
ast_test_validate(test,
actual_cache_entry == test_message1_1
|| actual_cache_entry == test_message2_1
|| actual_cache_entry == test_message2_2
|| actual_cache_entry == test_message2_3);
}
ao2_iterator_destroy(&i);
/* Check the remote cached items */
ao2_cleanup(cache_dump);
cache_dump = stasis_cache_dump_by_eid(cache, NULL, &foreign_eid1);
ast_test_validate(test, NULL != cache_dump);
ast_test_validate(test, 1 == ao2_container_count(cache_dump));
i = ao2_iterator_init(cache_dump, 0);
while ((obj = ao2_iterator_next(&i))) {
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
ast_test_validate(test, actual_cache_entry == test_message2_2);
}
ao2_iterator_destroy(&i);
/* Post snapshot 2 from a repeated eid. */
stasis_publish(topic, test_message2_4);
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "7"));
actual_len = consumer_wait_for(cache_consumer, 10);
ast_test_validate(test, 10 == actual_len);
actual_len = consumer_wait_for(topic_consumer, 10);
ast_test_validate(test, 10 == actual_len);
/* Check the cache */
ao2_cleanup(cache_dump);
cache_dump = stasis_cache_dump_all(cache, NULL);
ast_test_validate(test, NULL != cache_dump);
ast_test_validate(test, 4 == ao2_container_count(cache_dump));
i = ao2_iterator_init(cache_dump, 0);
while ((obj = ao2_iterator_next(&i))) {
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
ast_test_validate(test,
actual_cache_entry == test_message1_1
|| actual_cache_entry == test_message2_1
|| actual_cache_entry == test_message2_2
|| actual_cache_entry == test_message2_4);
}
ao2_iterator_destroy(&i);
/* Check all snapshot 2 cache entries. */
ao2_cleanup(cache_dump);
cache_dump = stasis_cache_get_all(cache, cache_type, "2");
ast_test_validate(test, NULL != cache_dump);
ast_test_validate(test, 3 == ao2_container_count(cache_dump));
i = ao2_iterator_init(cache_dump, 0);
while ((obj = ao2_iterator_next(&i))) {
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
ast_test_validate(test,
actual_cache_entry == test_message2_1
|| actual_cache_entry == test_message2_2
|| actual_cache_entry == test_message2_4);
}
ao2_iterator_destroy(&i);
/* Clear snapshot 1 */
test_message1_clear = stasis_cache_clear_create(test_message1_1);
ast_test_validate(test, NULL != test_message1_clear);
stasis_publish(topic, test_message1_clear);
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "1", NULL));
actual_len = consumer_wait_for(cache_consumer, 12);
ast_test_validate(test, 12 == actual_len);
actual_len = consumer_wait_for(topic_consumer, 11);
ast_test_validate(test, 11 == actual_len);
/* Check the cache */
ao2_cleanup(cache_dump);
cache_dump = stasis_cache_dump_all(cache, NULL);
ast_test_validate(test, NULL != cache_dump);
ast_test_validate(test, 3 == ao2_container_count(cache_dump));
i = ao2_iterator_init(cache_dump, 0);
while ((obj = ao2_iterator_next(&i))) {
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
ast_test_validate(test,
actual_cache_entry == test_message2_1
|| actual_cache_entry == test_message2_2
|| actual_cache_entry == test_message2_4);
}
ao2_iterator_destroy(&i);
/* Clear snapshot 2 from a remote eid */
test_message2_clear = stasis_cache_clear_create(test_message2_2);
ast_test_validate(test, NULL != test_message2_clear);
stasis_publish(topic, test_message2_clear);
ast_test_validate(test, check_cache_aggregate(cache, cache_type, "2", "5"));
actual_len = consumer_wait_for(cache_consumer, 14);
ast_test_validate(test, 14 == actual_len);
actual_len = consumer_wait_for(topic_consumer, 13);
ast_test_validate(test, 13 == actual_len);
/* Check the cache */
ao2_cleanup(cache_dump);
cache_dump = stasis_cache_dump_all(cache, NULL);
ast_test_validate(test, NULL != cache_dump);
ast_test_validate(test, 2 == ao2_container_count(cache_dump));
i = ao2_iterator_init(cache_dump, 0);
while ((obj = ao2_iterator_next(&i))) {
RAII_VAR(struct stasis_message *, actual_cache_entry, obj, ao2_cleanup);
ast_test_validate(test,
actual_cache_entry == test_message2_1
|| actual_cache_entry == test_message2_4);
}
ao2_iterator_destroy(&i);
return AST_TEST_PASS;
}
AST_TEST_DEFINE(router)
{
RAII_VAR(struct stasis_topic *, topic, NULL, ao2_cleanup);
@ -1399,6 +1755,7 @@ static int unload_module(void)
AST_TEST_UNREGISTER(cache_filter);
AST_TEST_UNREGISTER(cache);
AST_TEST_UNREGISTER(cache_dump);
AST_TEST_UNREGISTER(cache_eid_aggregate);
AST_TEST_UNREGISTER(router);
AST_TEST_UNREGISTER(router_cache_updates);
AST_TEST_UNREGISTER(interleaving);
@ -1423,6 +1780,7 @@ static int load_module(void)
AST_TEST_REGISTER(cache_filter);
AST_TEST_REGISTER(cache);
AST_TEST_REGISTER(cache_dump);
AST_TEST_REGISTER(cache_eid_aggregate);
AST_TEST_REGISTER(router);
AST_TEST_REGISTER(router_cache_updates);
AST_TEST_REGISTER(interleaving);

Write Preview
Loading…
Cancel
Save