Merge "endpoint/stasis: Eliminate duplicate events on endpoint status change"

2 files changed, 38 insertions, 16 deletions

diff --git a/main/endpoints.c b/main/endpoints.c
index 21326561c..b73edd379 100644
--- a/main/endpoints.c
+++ b/main/endpoints.c
@@ -303,13 +303,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(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;
@@ -325,9 +326,16 @@ static struct ast_endpoint *endpoint_internal_create(const char *tech, const cha
 
 		endpoint->tech_forward = stasis_forward_all(stasis_cp_single_topic(endpoint->topics),
 			stasis_cp_single_topic(tech_endpoint->topics));
+
 		endpoint_publish_snapshot(endpoint);
 		ao2_link(endpoints, endpoint);
 	} else {
+		endpoint->topics = stasis_cp_sink_create(ast_endpoint_cache_all(),
+			endpoint->id);
+		if (!endpoint->topics) {
+			return NULL;
+		}
+
 		ao2_link(tech_endpoints, endpoint);
 	}
 
diff --git a/main/stasis_cache_pattern.c b/main/stasis_cache_pattern.c
index bbe63ba1d..66563c4c6 100644
--- a/main/stasis_cache_pattern.c
+++ b/main/stasis_cache_pattern.c
@@ -138,20 +138,11 @@ struct stasis_cp_single *stasis_cp_single_create(struct stasis_cp_all *all,
 {
 	RAII_VAR(struct stasis_cp_single *, one, NULL, ao2_cleanup);
 
-	one = ao2_t_alloc(sizeof(*one), one_dtor, name);
+	one = stasis_cp_sink_create(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;
-	}
-
 	one->forward_topic_to_all = stasis_forward_all(one->topic, all->topic);
 	if (!one->forward_topic_to_all) {
 		return NULL;
@@ -166,6 +157,29 @@ struct stasis_cp_single *stasis_cp_single_create(struct stasis_cp_all *all,
 	return one;
 }
 
+struct stasis_cp_single *stasis_cp_sink_create(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;
+}
+
 void stasis_cp_single_unsubscribe(struct stasis_cp_single *one)
 {
 	if (!one) {