From: George Joseph Date: Wed, 27 May 2015 18:22:39 +0000 (-0600) Subject: endpoint/stasis: Eliminate duplicate events on endpoint status change X-Git-Tag: 13.8.0-rc1~173 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=3a1c4885be4eeb3f6c631b0cfb4709442c9464ae;p=thirdparty%2Fasterisk.git endpoint/stasis: Eliminate duplicate events on endpoint status change When an endpoint is created, its messages are forwarded to both the tech endpoint topic and the all endpoints topic. This is done so that various parties interested in endpoint messages can subscribe to just the tech endpoint and receive all messages associated with that particular technology, as opposed to subscribing to the all endpoints topic. Unfortunately, when the tech endpoint is created, it also forwards all of its messages to the all topic. This results in duplicate messages whenever an endpoint publishes its messages. This patch resolves the duplicate message issue by creating a new function for Stasis caching topics, stasis_cp_sink_create. In most respects, this acts as a normal caching topic, save that it no longer forwards messages it receives to the all endpoints topic. This allows it to act as an aggregation "sink", while preserving the necessary caching behaviour. ASTERISK-25137 #close Reported-by: Vitezslav Novy ASTERISK-25116 #close Reported-by: George Joseph Tested-by: George Joseph Change-Id: Ie47784adfb973ab0063e59fc18f390d7dd26d17b --- diff --git a/include/asterisk/stasis_cache_pattern.h b/include/asterisk/stasis_cache_pattern.h index 2ea643e192..27761351ad 100644 --- a/include/asterisk/stasis_cache_pattern.h +++ b/include/asterisk/stasis_cache_pattern.h @@ -109,6 +109,8 @@ struct stasis_cp_single; /*! * \brief Create the 'one' side of the cache pattern. * + * Create the 'one' and forward to all's topic and topic_cached. + * * Dispose of using stasis_cp_single_unsubscribe(). * * \param all Corresponding all side. @@ -118,6 +120,32 @@ struct stasis_cp_single; struct stasis_cp_single *stasis_cp_single_create(struct stasis_cp_all *all, const char *name); +/*! + * \brief Create the 'one' side of the cache pattern. + * + * Create the 'one' but do not automatically forward. + * + * Dispose of using stasis_cp_single_unsubscribe(). + * + * \param all Corresponding all side. + * \param name Base name for the topics. + * \return One side instance + */ +struct stasis_cp_single *stasis_cp_single_create_only(struct stasis_cp_all *all, + const char *name); + +/*! + * \brief Set up a topic and topic cache forward. + * + * Forward 'from' to 'to'. + * + * \param from Source 'one' side instance. + * \param to Destination 'one' side instance. + * \retval 0 Success + * \retval -1 Failure + */ +int stasis_cp_single_forward(struct stasis_cp_single *from, struct stasis_cp_single *to); + /*! * \brief Stops caching and forwarding messages. * diff --git a/main/endpoints.c b/main/endpoints.c index 0155adfebd..8f3ae366e7 100644 --- a/main/endpoints.c +++ b/main/endpoints.c @@ -74,8 +74,6 @@ struct ast_endpoint { struct stasis_message_router *router; /*! ast_str_container of channels associated with this endpoint */ struct ao2_container *channel_ids; - /*! Forwarding subscription from an endpoint to its tech endpoint */ - struct stasis_forward *tech_forward; }; static int endpoint_hash(const void *obj, int flags) @@ -303,13 +301,14 @@ static struct ast_endpoint *endpoint_internal_create(const char *tech, const cha return NULL; } - endpoint->topics = stasis_cp_single_create(ast_endpoint_cache_all(), - endpoint->id); - if (!endpoint->topics) { - return NULL; - } - if (!ast_strlen_zero(resource)) { + + endpoint->topics = stasis_cp_single_create_only(ast_endpoint_cache_all(), + endpoint->id); + if (!endpoint->topics) { + return NULL; + } + endpoint->router = stasis_message_router_create_pool(ast_endpoint_topic(endpoint)); if (!endpoint->router) { return NULL; @@ -323,11 +322,19 @@ static struct ast_endpoint *endpoint_internal_create(const char *tech, const cha return NULL; } - endpoint->tech_forward = stasis_forward_all(stasis_cp_single_topic(endpoint->topics), - stasis_cp_single_topic(tech_endpoint->topics)); + if (stasis_cp_single_forward(endpoint->topics, tech_endpoint->topics)) { + return NULL; + } + endpoint_publish_snapshot(endpoint); ao2_link(endpoints, endpoint); } else { + endpoint->topics = stasis_cp_single_create(ast_endpoint_cache_all(), + endpoint->id); + if (!endpoint->topics) { + return NULL; + } + ao2_link(tech_endpoints, endpoint); } @@ -375,7 +382,6 @@ void ast_endpoint_shutdown(struct ast_endpoint *endpoint) } ao2_unlink(endpoints, endpoint); - endpoint->tech_forward = stasis_forward_cancel(endpoint->tech_forward); clear_msg = create_endpoint_snapshot_message(endpoint); if (clear_msg) { diff --git a/main/stasis_cache_pattern.c b/main/stasis_cache_pattern.c index 9e3de367ad..ccc9ebf081 100644 --- a/main/stasis_cache_pattern.c +++ b/main/stasis_cache_pattern.c @@ -138,6 +138,30 @@ struct stasis_cp_single *stasis_cp_single_create(struct stasis_cp_all *all, { RAII_VAR(struct stasis_cp_single *, one, NULL, ao2_cleanup); + one = stasis_cp_single_create_only(all, name); + if (!one) { + return NULL; + } + + one->forward_topic_to_all = stasis_forward_all(one->topic, all->topic); + if (!one->forward_topic_to_all) { + return NULL; + } + one->forward_cached_to_all = stasis_forward_all( + stasis_caching_get_topic(one->topic_cached), all->topic_cached); + if (!one->forward_cached_to_all) { + return NULL; + } + + ao2_ref(one, +1); + return one; +} + +struct stasis_cp_single *stasis_cp_single_create_only(struct stasis_cp_all *all, + const char *name) +{ + RAII_VAR(struct stasis_cp_single *, one, NULL, ao2_cleanup); + one = ao2_t_alloc(sizeof(*one), one_dtor, name); if (!one) { return NULL; @@ -152,18 +176,25 @@ struct stasis_cp_single *stasis_cp_single_create(struct stasis_cp_all *all, return NULL; } - one->forward_topic_to_all = stasis_forward_all(one->topic, all->topic); - if (!one->forward_topic_to_all) { - return NULL; + ao2_ref(one, +1); + return one; +} + +int stasis_cp_single_forward(struct stasis_cp_single *from, struct stasis_cp_single *to) +{ + from->forward_topic_to_all = stasis_forward_all(from->topic, to->topic); + if (!from->forward_topic_to_all) { + return -1;; } - one->forward_cached_to_all = stasis_forward_all( - stasis_caching_get_topic(one->topic_cached), all->topic_cached); - if (!one->forward_cached_to_all) { - return NULL; + + from->forward_cached_to_all = stasis_forward_all( + stasis_caching_get_topic(from->topic_cached), + stasis_caching_get_topic(to->topic_cached)); + if (!from->forward_cached_to_all) { + return -1; } - ao2_ref(one, +1); - return one; + return 0; } void stasis_cp_single_unsubscribe(struct stasis_cp_single *one)