Split caching out from the stasis_caching_topic.

In working with res_stasis, I discovered a significant limitation to
the current structure of stasis_caching_topics: you cannot subscribe
to cache updates for a single channel/bridge/endpoint/etc.

To address this, this patch splits the cache away from the
stasis_caching_topic, making it a first class object. The stasis_cache
object is shared amongst individual stasis_caching_topics that are
created per channel/endpoint/etc. These are still forwarded to global
whatever_all_cached topics, so their use from most of the code does
not change.

In making these changes, I noticed that we frequently used a similar
pattern for bridges, endpoints and channels:

     single_topic  ---------------->  all_topic
           ^
           |
     single_topic_cached  ----+---->  all_topic_cached
                              |
                              +---->  cache

This pattern was extracted as the 'Stasis Caching Pattern', defined in
stasis_caching_pattern.h. This avoids a lot of duplicate code between
the different domain objects.

Since the cache is now disassociated from its upstream caching topics,
this also necessitated a change to how the 'guaranteed' flag worked
for retrieving from a cache. The code for handling the caching
guarantee was extracted into a 'stasis_topic_wait' function, which
works for any stasis_topic.

(closes issue ASTERISK-22002)
Review: https://reviewboard.asterisk.org/r/2672/


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@395954 65c4cc65-6c06-0410-ace0-fbb531ad65f3

1 files changed, 31 insertions, 13 deletions

diff --git a/main/channel_internal_api.c b/main/channel_internal_api.c
index a1d20871d..35bcb187d 100644
--- a/main/channel_internal_api.c
+++ b/main/channel_internal_api.c
@@ -44,6 +44,7 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 #include "asterisk/data.h"
 #include "asterisk/endpoints.h"
 #include "asterisk/indications.h"
+#include "asterisk/stasis_cache_pattern.h"
 #include "asterisk/stasis_channels.h"
 #include "asterisk/stasis_endpoints.h"
 #include "asterisk/stringfields.h"
@@ -208,7 +209,7 @@ struct ast_channel {
 	char dtmf_digit_to_emulate;			/*!< Digit being emulated */
 	char sending_dtmf_digit;			/*!< Digit this channel is currently sending out. (zero if not sending) */
 	struct timeval sending_dtmf_tv;		/*!< The time this channel started sending the current digit. (Invalid if sending_dtmf_digit is zero.) */
-	struct stasis_topic *topic;			/*!< Topic for all channel's events */
+	struct stasis_cp_single *topics;		/*!< Topic for all channel's events */
 	struct stasis_subscription *forwarder;		/*!< Subscription for event forwarding to all topic */
 	struct stasis_subscription *endpoint_forward;	/*!< Subscription for event forwarding to endpoint's topic */
 };
@@ -1434,8 +1435,8 @@ void ast_channel_internal_cleanup(struct ast_channel *chan)
 	chan->forwarder = stasis_unsubscribe(chan->forwarder);
 	chan->endpoint_forward = stasis_unsubscribe(chan->endpoint_forward);
 
-	ao2_cleanup(chan->topic);
-	chan->topic = NULL;
+	stasis_cp_single_unsubscribe(chan->topics);
+	chan->topics = NULL;
 }
 
 void ast_channel_internal_finalize(struct ast_channel *chan)
@@ -1450,16 +1451,31 @@ int ast_channel_internal_is_finalized(struct ast_channel *chan)
 
 struct stasis_topic *ast_channel_topic(struct ast_channel *chan)
 {
-	return chan ? chan->topic : ast_channel_topic_all();
+	if (!chan) {
+		return ast_channel_topic_all();
+	}
+
+	return stasis_cp_single_topic(chan->topics);
 }
 
-int ast_channel_forward_endpoint(struct ast_channel *chan, struct ast_endpoint *endpoint)
+struct stasis_topic *ast_channel_topic_cached(struct ast_channel *chan)
+{
+	if (!chan) {
+		return ast_channel_topic_all_cached();
+	}
+
+	return stasis_cp_single_topic_cached(chan->topics);
+}
+
+int ast_channel_forward_endpoint(struct ast_channel *chan,
+	struct ast_endpoint *endpoint)
 {
 	ast_assert(chan != NULL);
 	ast_assert(endpoint != NULL);
 
 	chan->endpoint_forward =
-		stasis_forward_all(chan->topic, ast_endpoint_topic(endpoint));
+		stasis_forward_all(ast_channel_topic(chan),
+			ast_endpoint_topic(endpoint));
 
 	if (chan->endpoint_forward == NULL) {
 		return -1;
@@ -1468,19 +1484,21 @@ int ast_channel_forward_endpoint(struct ast_channel *chan, struct ast_endpoint *
 	return 0;
 }
 
-void ast_channel_internal_setup_topics(struct ast_channel *chan)
+int ast_channel_internal_setup_topics(struct ast_channel *chan)
 {
 	const char *topic_name = chan->uniqueid;
-	ast_assert(chan->topic == NULL);
-	ast_assert(chan->forwarder == NULL);
+	ast_assert(chan->topics == NULL);
 
 	if (ast_strlen_zero(topic_name)) {
 		topic_name = "<dummy-channel>";
 	}
 
-	chan->topic = stasis_topic_create(topic_name);
-	chan->forwarder = stasis_forward_all(chan->topic, ast_channel_topic_all());
+	chan->topics = stasis_cp_single_create(
+		ast_channel_cache_all(), topic_name);
 
-	ast_assert(chan->topic != NULL);
-	ast_assert(chan->forwarder != NULL);
+	if (!chan->topics) {
+		return -1;
+	}
+
+	return 0;
 }