diff options
author | George Joseph <george.joseph@fairview5.com> | 2015-05-27 12:22:39 -0600 |
---|---|---|
committer | George Joseph <george.joseph@fairview5.com> | 2015-05-27 17:16:33 -0500 |
commit | 6fca75bb628dfff2ab112e80b0228cf3ac0b8a05 (patch) | |
tree | 3f33d4ba544fc241c54d0120cb7f78fed6889330 /main | |
parent | b8ac68382225e2c0227d4764eebd7fdcea93249f (diff) |
endpoint/stasis: Eliminate duplicate events on endpoint status change
When an endpoint was created, it's messages were being forwarded to
both the tech endpoint topic and the all endpoints topic. Since
the tech topic was also forwarded to all, this was resulting in
duplicate messages whenever an endpoint published. This patch
causes the endpoint to only forward to the tech topic and lets
the tech topic forward to all.
To accomplish this, the existing stasis_cp_single_create function
(which both creates and forwards) was cloned and split into 2
functions, one that creates the topic and one that sets up the
forwarding. This allows endpoint_internal_create to create
the topic from the endpoint_all cache without forwarding it there,
then allows it to do the forward to the tech's topic.
ASTERISK-25137 #close
Reported-by: Vitezslav Novy
ASTERISK-25116 #close
Reported-by: George Joseph <george.joseph@fairview5.com>
Tested-by: George Joseph <george.joseph@fairview5.com>
Change-Id: I26d7d4926a0861748fd3bdffe316b75b549a801c
Diffstat (limited to 'main')
-rw-r--r-- | main/endpoints.c | 28 | ||||
-rw-r--r-- | main/stasis_cache_pattern.c | 49 |
2 files changed, 57 insertions, 20 deletions
diff --git a/main/endpoints.c b/main/endpoints.c index df9d289c7..979d43e43 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 bbe63ba1d..d916a48b1 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) |