diff options
| -rw-r--r-- | include/asterisk/devicestate.h | 21 | ||||
| -rw-r--r-- | include/asterisk/stasis.h | 258 | ||||
| -rw-r--r-- | main/app.c | 15 | ||||
| -rw-r--r-- | main/devicestate.c | 327 | ||||
| -rw-r--r-- | main/stasis_cache.c | 650 | ||||
| -rw-r--r-- | main/stasis_message.c | 30 | ||||
| -rw-r--r-- | tests/test_devicestate.c | 141 | ||||
| -rw-r--r-- | tests/test_stasis.c | 396 |
8 files changed, 1459 insertions, 379 deletions
diff --git a/include/asterisk/devicestate.h b/include/asterisk/devicestate.h index e724588af..ccc46311e 100644 --- a/include/asterisk/devicestate.h +++ b/include/asterisk/devicestate.h @@ -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]; }; /*! diff --git a/include/asterisk/stasis.h b/include/asterisk/stasis.h index 1a420da2f..20870e6d6 100644 --- a/include/asterisk/stasis.h +++ b/include/asterisk/stasis.h @@ -290,13 +290,51 @@ 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 * \return Type of \a msg @@ -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 @@ -661,6 +702,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. * * This is the backend store for a \ref stasis_caching_topic. The cache is @@ -669,13 +805,41 @@ 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. * * The idea is that some topics publish 'snapshots' of some other object's state @@ -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); /*! @} */ diff --git a/main/app.c b/main/app.c index 6e1a176c6..ce0d16cdd 100644 --- a/main/app.c +++ b/main/app.c @@ -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; diff --git a/main/devicestate.c b/main/devicestate.c index 1f0c968aa..5c2340863 100644 --- a/main/devicestate.c +++ b/main/devicestate.c @@ -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; } diff --git a/main/stasis_cache.c b/main/stasis_cache.c index 1c5053915..22ace60af 100644 --- a/main/stasis_cache.c +++ b/main/stasis_cache.c @@ -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) diff --git a/main/stasis_message.c b/main/stasis_message.c index 240845aff..1db2ae97a 100644 --- a/main/stasis_message.c +++ b/main/stasis_message.c @@ -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) { diff --git a/tests/test_devicestate.c b/tests/test_devicestate.c index 5a3d255d1..f7075b426 100644 --- a/tests/test_devicestate.c +++ b/tests/test_devicestate.c @@ -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; } diff --git a/tests/test_stasis.c b/tests/test_stasis.c index b2f4777e7..4c042c05b 100644 --- a/tests/test_stasis.c +++ b/tests/test_stasis.c @@ -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); |