[Asterisk-code-review] endpoint/stasis: Eliminate duplicate events on endpoint stat... (asterisk[13])
Matt Jordan
asteriskteam at digium.com
Mon Dec 28 15:11:14 CST 2015
Matt Jordan has submitted this change and it was merged.
Change subject: endpoint/stasis: Eliminate duplicate events on endpoint status change
......................................................................
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 <george.joseph at fairview5.com>
Tested-by: George Joseph <george.joseph at fairview5.com>
Change-Id: Ie47784adfb973ab0063e59fc18f390d7dd26d17b
---
M include/asterisk/stasis_cache_pattern.h
M main/endpoints.c
M main/stasis_cache_pattern.c
3 files changed, 86 insertions(+), 21 deletions(-)
Approvals:
Matt Jordan: Looks good to me, approved; Verified
Joshua Colp: Looks good to me, but someone else must approve
diff --git a/include/asterisk/stasis_cache_pattern.h b/include/asterisk/stasis_cache_pattern.h
index 2ea643e..2776135 100644
--- a/include/asterisk/stasis_cache_pattern.h
+++ b/include/asterisk/stasis_cache_pattern.h
@@ -109,6 +109,8 @@
/*!
* \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.
@@ -119,6 +121,32 @@
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.
*
* \param one One side of the cache pattern.
diff --git a/main/endpoints.c b/main/endpoints.c
index 0155adf..8f3ae36 100644
--- a/main/endpoints.c
+++ b/main/endpoints.c
@@ -74,8 +74,6 @@
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 @@
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 @@
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 @@
}
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 9e3de36..ccc9ebf 100644
--- a/main/stasis_cache_pattern.c
+++ b/main/stasis_cache_pattern.c
@@ -138,17 +138,8 @@
{
RAII_VAR(struct stasis_cp_single *, one, NULL, ao2_cleanup);
- one = ao2_t_alloc(sizeof(*one), one_dtor, name);
+ one = stasis_cp_single_create_only(all, name);
if (!one) {
- return NULL;
- }
-
- one->topic = stasis_topic_create(name);
- if (!one->topic) {
- return NULL;
- }
- one->topic_cached = stasis_caching_topic_create(one->topic, all->cache);
- if (!one->topic_cached) {
return NULL;
}
@@ -166,6 +157,46 @@
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;
+ }
+
+ one->topic = stasis_topic_create(name);
+ if (!one->topic) {
+ return NULL;
+ }
+ one->topic_cached = stasis_caching_topic_create(one->topic, all->cache);
+ if (!one->topic_cached) {
+ 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;;
+ }
+
+ 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;
+ }
+
+ return 0;
+}
+
void stasis_cp_single_unsubscribe(struct stasis_cp_single *one)
{
if (!one) {
--
To view, visit https://gerrit.asterisk.org/570
To unsubscribe, visit https://gerrit.asterisk.org/settings
Gerrit-MessageType: merged
Gerrit-Change-Id: Ie47784adfb973ab0063e59fc18f390d7dd26d17b
Gerrit-PatchSet: 11
Gerrit-Project: asterisk
Gerrit-Branch: 13
Gerrit-Owner: George Joseph <george.joseph at fairview5.com>
Gerrit-Reviewer: Anonymous Coward #1000019
Gerrit-Reviewer: Ashley Sanders <asanders at digium.com>
Gerrit-Reviewer: David M. Lee <dlee at digium.com>
Gerrit-Reviewer: George Joseph <george.joseph at fairview5.com>
Gerrit-Reviewer: Joshua Colp <jcolp at digium.com>
Gerrit-Reviewer: Matt Jordan <mjordan at digium.com>
More information about the asterisk-code-review
mailing list