[Asterisk-code-review] RFC: endpoint/stasis: Eliminate duplicate events on endpoint... (asterisk[13])
George Joseph
asteriskteam at digium.com
Tue Jun 2 13:32:17 CDT 2015
George Joseph has uploaded a new change for review.
https://gerrit.asterisk.org/570
Change subject: RFC: endpoint/stasis: Eliminate duplicate events on endpoint status change
......................................................................
RFC: endpoint/stasis: Eliminate duplicate events on endpoint status change
THIS IS AN RFC PATCH ONLY. DO NOT COMMIT
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.
Since there was a test looking for a cache removal that now
doesn't exist, that test has been disabled and an error
message in stasis_cache.c has been disabled. The final
patch will remove this code.
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.c
M main/stasis_cache_pattern.c
M tests/test_stasis_endpoints.c
5 files changed, 92 insertions(+), 25 deletions(-)
git pull ssh://gerrit.asterisk.org:29418/asterisk refs/changes/70/570/1
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 ce0ab02..a640014 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.c b/main/stasis_cache.c
index 9129c00..339f9cf 100644
--- a/main/stasis_cache.c
+++ b/main/stasis_cache.c
@@ -841,13 +841,15 @@
stasis_publish(caching_topic->topic, update);
}
ao2_cleanup(update);
- } else {
+ }
+#if 0
+ 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(msg_type), msg_id);
}
-
+#endif
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,
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) {
diff --git a/tests/test_stasis_endpoints.c b/tests/test_stasis_endpoints.c
index 7ac5291..310a90e 100644
--- a/tests/test_stasis_endpoints.c
+++ b/tests/test_stasis_endpoints.c
@@ -182,7 +182,7 @@
ast_endpoint_shutdown(uut);
uut = NULL;
-
+#if 0
/* Note: there's a few messages between the creation and the clear.
* Wait for all of them... */
message_index = stasis_message_sink_wait_for(sink, message_index + 4,
@@ -199,7 +199,7 @@
ast_test_validate(test,
0 == strcmp(__func__, actual_snapshot->resource));
ast_test_validate(test, NULL == update->new_snapshot);
-
+#endif
return AST_TEST_PASS;
}
--
To view, visit https://gerrit.asterisk.org/570
To unsubscribe, visit https://gerrit.asterisk.org/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: Ie47784adfb973ab0063e59fc18f390d7dd26d17b
Gerrit-PatchSet: 1
Gerrit-Project: asterisk
Gerrit-Branch: 13
Gerrit-Owner: George Joseph <george.joseph at fairview5.com>
More information about the asterisk-code-review
mailing list