diff --git a/include/asterisk/stasis_cache_pattern.h b/include/asterisk/stasis_cache_pattern.h index 2ea643e192..e61d3e931c 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,23 @@ struct stasis_cp_single; struct stasis_cp_single *stasis_cp_single_create(struct stasis_cp_all *all, const char *name); +/*! + * \brief Create a sink in the cache pattern + * + * Create the 'one' but do not automatically forward to the all's topic. + * This is useful when aggregating other topic's messages created with + * \c stasis_cp_single_create in another caching topic without replicating + * those messages in the all's topics. + * + * 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_sink_create(struct stasis_cp_all *all, + const char *name); + /*! * \brief Stops caching and forwarding messages. * diff --git a/main/endpoints.c b/main/endpoints.c index 21326561ce..b73edd379a 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 bbe63ba1de..66563c4c6d 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_sink_create(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_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; @@ -152,16 +176,6 @@ 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; - } - 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; }