diff options
-rw-r--r-- | include/asterisk/channel.h | 13 | ||||
-rw-r--r-- | include/asterisk/stasis.h | 18 | ||||
-rw-r--r-- | include/asterisk/stasis_endpoints.h | 4 | ||||
-rw-r--r-- | main/channel_internal_api.c | 5 | ||||
-rw-r--r-- | main/endpoints.c | 34 | ||||
-rw-r--r-- | main/stasis_cache.c | 78 | ||||
-rw-r--r-- | main/stasis_endpoints.c | 6 | ||||
-rw-r--r-- | res/stasis_http/resource_endpoints.c | 2 |
8 files changed, 126 insertions, 34 deletions
diff --git a/include/asterisk/channel.h b/include/asterisk/channel.h index c2edc18e7..42d50f21b 100644 --- a/include/asterisk/channel.h +++ b/include/asterisk/channel.h @@ -152,6 +152,7 @@ extern "C" { #include "asterisk/framehook.h" #include "asterisk/stasis.h" #include "asterisk/json.h" +#include "asterisk/endpoints.h" #define DATASTORE_INHERIT_FOREVER INT_MAX @@ -4265,4 +4266,16 @@ struct ast_channel *ast_channel_yank(struct ast_channel *yankee); */ int ast_channel_move(struct ast_channel *dest, struct ast_channel *source); +/*! + * \since 12 + * \brief Forward channel stasis messages to the given endpoint + * + * \param chan The channel to forward from + * \param endpoint The endpoint to forward to + * + * \retval 0 Success + * \retval non-zero Failure + */ +int ast_channel_forward_endpoint(struct ast_channel *chan, struct ast_endpoint *endpoint); + #endif /* _ASTERISK_CHANNEL_H */ diff --git a/include/asterisk/stasis.h b/include/asterisk/stasis.h index bd303ebd8..19ebb41c9 100644 --- a/include/asterisk/stasis.h +++ b/include/asterisk/stasis.h @@ -670,9 +670,23 @@ struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *cachi * \return \c NULL if message is not found. * \since 12 */ -struct stasis_message *stasis_cache_get(struct stasis_caching_topic *caching_topic, +#define stasis_cache_get(caching_topic, type, id) stasis_cache_get_extended(caching_topic, type, id, 0) + +/*! + * \brief Retrieve an item from the cache. + * \param caching_topic The topic returned from stasis_caching_topic_create(). + * \param type Type of message to retrieve. + * \param id Identity of the snapshot to retrieve. + * \param guaranteed If set to 1 it is guaranteed that any pending messages have been processed. + * \return Message from the cache. The cache still owns the message, so + * ao2_ref() if you want to keep it. + * \return \c NULL if message is not found. + * \since 12 + */ +struct stasis_message *stasis_cache_get_extended(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, - const char *id); + const char *id, + unsigned int guaranteed); /*! * \brief Dump cached items to a subscription diff --git a/include/asterisk/stasis_endpoints.h b/include/asterisk/stasis_endpoints.h index 2aeb614e6..10abcd77b 100644 --- a/include/asterisk/stasis_endpoints.h +++ b/include/asterisk/stasis_endpoints.h @@ -122,12 +122,14 @@ struct stasis_caching_topic *ast_endpoint_topic_all_cached(void); * * \param tech Name of the endpoint's technology. * \param resource Resource name of the endpoint. + * \param guaranteed Whether to require all pending messages to have been processed or not. * \return Snapshot of the endpoint with the given name. * \return \c NULL if endpoint is not found, or on error. * \since 12 */ struct ast_endpoint_snapshot *ast_endpoint_latest_snapshot(const char *tech, - const char *resource + const char *resource, + unsigned int guaranteed ); /*! @} */ diff --git a/main/channel_internal_api.c b/main/channel_internal_api.c index 4e5284c55..1e21cced0 100644 --- a/main/channel_internal_api.c +++ b/main/channel_internal_api.c @@ -1406,8 +1406,7 @@ struct stasis_topic *ast_channel_topic(struct ast_channel *chan) return chan ? chan->topic : ast_channel_topic_all(); } -int ast_endpoint_add_channel(struct ast_endpoint *endpoint, - struct ast_channel *chan) +int ast_channel_forward_endpoint(struct ast_channel *chan, struct ast_endpoint *endpoint) { ast_assert(chan != NULL); ast_assert(endpoint != NULL); @@ -1419,8 +1418,6 @@ int ast_endpoint_add_channel(struct ast_endpoint *endpoint, return -1; } - ast_publish_channel_state(chan); - return 0; } diff --git a/main/endpoints.c b/main/endpoints.c index 9f6d2c062..d689f2e6e 100644 --- a/main/endpoints.c +++ b/main/endpoints.c @@ -122,30 +122,23 @@ static void endpoint_dtor(void *obj) ast_string_field_free_memory(endpoint); } -static void endpoint_channel_snapshot(void *data, - struct stasis_subscription *sub, struct stasis_topic *topic, - struct stasis_message *message) -{ - struct ast_endpoint *endpoint = data; - struct ast_channel_snapshot *snapshot = stasis_message_data(message); - RAII_VAR(char *, existing_id, NULL, ao2_cleanup); - int publish = 0; +int ast_endpoint_add_channel(struct ast_endpoint *endpoint, + struct ast_channel *chan) +{ + ast_assert(chan != NULL); ast_assert(endpoint != NULL); - ast_assert(snapshot != NULL); + + ast_channel_forward_endpoint(chan, endpoint); ao2_lock(endpoint); - existing_id = ao2_find(endpoint->channel_ids, snapshot->uniqueid, - OBJ_POINTER); - if (!existing_id) { - ast_str_container_add(endpoint->channel_ids, - snapshot->uniqueid); - publish = 1; - } + ast_str_container_add(endpoint->channel_ids, ast_channel_uniqueid(chan)); ao2_unlock(endpoint); - if (publish) { - endpoint_publish_snapshot(endpoint); - } + + ast_publish_channel_state(chan); + endpoint_publish_snapshot(endpoint); + + return 0; } /*! \brief Handler for channel snapshot cache clears */ @@ -237,9 +230,6 @@ struct ast_endpoint *ast_endpoint_create(const char *tech, const char *resource) return NULL; } r |= stasis_message_router_add(endpoint->router, - ast_channel_snapshot_type(), endpoint_channel_snapshot, - endpoint); - r |= stasis_message_router_add(endpoint->router, stasis_cache_clear_type(), endpoint_cache_clear, endpoint); r |= stasis_message_router_set_default(endpoint->router, diff --git a/main/stasis_cache.c b/main/stasis_cache.c index 546ad4998..115bb7b67 100644 --- a/main/stasis_cache.c +++ b/main/stasis_cache.c @@ -46,10 +46,13 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$") struct stasis_caching_topic { struct ao2_container *cache; struct stasis_topic *topic; + struct stasis_topic *original_topic; struct stasis_subscription *sub; snapshot_get_id id_fn; }; +static struct stasis_message_type *cache_guarantee_type(void); + static void stasis_caching_topic_dtor(void *obj) { struct stasis_caching_topic *caching_topic = obj; ast_assert(!stasis_subscription_is_subscribed(caching_topic->sub)); @@ -60,6 +63,8 @@ static void stasis_caching_topic_dtor(void *obj) { caching_topic->cache = NULL; ao2_cleanup(caching_topic->topic); caching_topic->topic = NULL; + ao2_cleanup(caching_topic->original_topic); + caching_topic->original_topic = NULL; } struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *caching_topic) @@ -204,13 +209,62 @@ static struct stasis_message *cache_put(struct stasis_caching_topic *caching_top return old_snapshot; } -struct stasis_message *stasis_cache_get(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, const char *id) +/*! \internal */ +struct caching_guarantee { + ast_mutex_t lock; + ast_cond_t cond; + unsigned int done:1; +}; + +static void caching_guarantee_dtor(void *obj) +{ + struct caching_guarantee *guarantee = obj; + + ast_assert(guarantee->done == 1); + + ast_mutex_destroy(&guarantee->lock); + ast_cond_destroy(&guarantee->cond); +} + +static struct stasis_message *caching_guarantee_create(void) +{ + RAII_VAR(struct caching_guarantee *, guarantee, NULL, ao2_cleanup); + RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup); + + if (!(guarantee = ao2_alloc(sizeof(*guarantee), caching_guarantee_dtor))) { + return NULL; + } + + ast_mutex_init(&guarantee->lock); + ast_cond_init(&guarantee->cond, NULL); + + if (!(msg = stasis_message_create(cache_guarantee_type(), guarantee))) { + return NULL; + } + + ao2_ref(msg, +1); + return msg; +} + +struct stasis_message *stasis_cache_get_extended(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, const char *id, unsigned int guaranteed) { RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup); RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup); ast_assert(caching_topic->cache != NULL); + if (guaranteed) { + RAII_VAR(struct stasis_message *, msg, caching_guarantee_create(), ao2_cleanup); + struct caching_guarantee *guarantee = stasis_message_data(msg); + + ast_mutex_lock(&guarantee->lock); + stasis_publish(caching_topic->original_topic, msg); + while (!guarantee->done) { + ast_cond_wait(&guarantee->cond, &guarantee->lock); + } + ast_mutex_unlock(&guarantee->lock); + } + search_entry = cache_entry_create(type, id, NULL); if (search_entry == NULL) { return NULL; @@ -261,6 +315,7 @@ struct ao2_container *stasis_cache_dump(struct stasis_caching_topic *caching_top STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type); STASIS_MESSAGE_TYPE_DEFN(stasis_cache_update_type); +STASIS_MESSAGE_TYPE_DEFN(cache_guarantee_type); struct stasis_message *stasis_cache_clear_create(struct stasis_message *id_message) { @@ -340,6 +395,18 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub, stru caching_topic_needs_unref = caching_topic; } + /* Handle cache guarantee event */ + if (cache_guarantee_type() == stasis_message_type(message)) { + struct caching_guarantee *guarantee = stasis_message_data(message); + + ast_mutex_lock(&guarantee->lock); + guarantee->done = 1; + ast_cond_signal(&guarantee->cond); + ast_mutex_unlock(&guarantee->lock); + + return; + } + /* Handle cache clear event */ if (stasis_cache_clear_type() == stasis_message_type(message)) { RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup); @@ -423,6 +490,10 @@ struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *or if (sub == NULL) { return NULL; } + + ao2_ref(original_topic, +1); + caching_topic->original_topic = original_topic; + /* This is for the reference contained in the subscription above */ ao2_ref(caching_topic, +1); caching_topic->sub = sub; @@ -435,6 +506,7 @@ static void stasis_cache_cleanup(void) { STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_clear_type); STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_update_type); + STASIS_MESSAGE_TYPE_CLEANUP(cache_guarantee_type); } int stasis_cache_init(void) @@ -449,6 +521,10 @@ int stasis_cache_init(void) return -1; } + if (STASIS_MESSAGE_TYPE_INIT(cache_guarantee_type) != 0) { + return -1; + } + return 0; } diff --git a/main/stasis_endpoints.c b/main/stasis_endpoints.c index a7b7f556b..d5347cbcb 100644 --- a/main/stasis_endpoints.c +++ b/main/stasis_endpoints.c @@ -52,7 +52,7 @@ struct stasis_caching_topic *ast_endpoint_topic_all_cached(void) } struct ast_endpoint_snapshot *ast_endpoint_latest_snapshot(const char *tech, - const char *name) + const char *name, unsigned int guaranteed) { RAII_VAR(char *, id, NULL, ast_free); RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup); @@ -63,8 +63,8 @@ struct ast_endpoint_snapshot *ast_endpoint_latest_snapshot(const char *tech, return NULL; } - msg = stasis_cache_get(ast_endpoint_topic_all_cached(), - ast_endpoint_snapshot_type(), id); + msg = stasis_cache_get_extended(ast_endpoint_topic_all_cached(), + ast_endpoint_snapshot_type(), id, guaranteed); if (!msg) { return NULL; } diff --git a/res/stasis_http/resource_endpoints.c b/res/stasis_http/resource_endpoints.c index 52d05c093..306413d75 100644 --- a/res/stasis_http/resource_endpoints.c +++ b/res/stasis_http/resource_endpoints.c @@ -140,7 +140,7 @@ void stasis_http_get_endpoint(struct ast_variable *headers, RAII_VAR(struct ast_json *, json, NULL, ast_json_unref); RAII_VAR(struct ast_endpoint_snapshot *, snapshot, NULL, ao2_cleanup); - snapshot = ast_endpoint_latest_snapshot(args->tech, args->resource); + snapshot = ast_endpoint_latest_snapshot(args->tech, args->resource, 0); if (!snapshot) { stasis_http_response_error(response, 404, "Not Found", "Endpoint not found"); |