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

11 files changed, 366 insertions, 54 deletions

diff --git a/include/asterisk/app.h b/include/asterisk/app.h
index 91438a2d0..251288546 100644
--- a/include/asterisk/app.h
+++ b/include/asterisk/app.h
@@ -1251,7 +1251,13 @@ struct stasis_topic *ast_mwi_topic(const char *uniqueid);
  * \retval NULL if it has not been allocated
  * \since 12
  */
-struct stasis_caching_topic *ast_mwi_topic_cached(void);
+struct stasis_topic *ast_mwi_topic_cached(void);
+
+/*!
+ * \brief Backend cache for ast_mwi_topic_cached().
+ * \retval Cache of \ref ast_mwi_state.
+ */
+struct stasis_cache *ast_mwi_state_cache(void);
 
 /*!
  * \brief Get the \ref stasis message type for MWI messages
diff --git a/include/asterisk/bridge.h b/include/asterisk/bridge.h
index a920a74e4..5f22d37ef 100644
--- a/include/asterisk/bridge.h
+++ b/include/asterisk/bridge.h
@@ -277,6 +277,8 @@ struct ast_bridge {
 	struct ast_bridge_technology *technology;
 	/*! Private information unique to the bridge technology */
 	void *tech_pvt;
+	/*! Per-bridge topics */
+	struct stasis_cp_single *topics;
 	/*! Call ID associated with the bridge */
 	struct ast_callid *callid;
 	/*! Linked list of channels participating in the bridge */
diff --git a/include/asterisk/channel.h b/include/asterisk/channel.h
index f6cdd4e62..282230a02 100644
--- a/include/asterisk/channel.h
+++ b/include/asterisk/channel.h
@@ -4186,6 +4186,21 @@ struct varshead *ast_channel_get_vars(struct ast_channel *chan);
 struct stasis_topic *ast_channel_topic(struct ast_channel *chan);
 
 /*!
+ * \since 12
+ * \brief A topic which publishes the events for a particular channel.
+ *
+ * \ref ast_channel_snapshot messages are replaced with \ref stasis_cache_update
+ *
+ * If the given \a chan is \c NULL, ast_channel_topic_all_cached() is returned.
+ *
+ * \param chan Channel, or \c NULL.
+ *
+ * \retval Topic for channel's events.
+ * \retval ast_channel_topic_all() if \a chan is \c NULL.
+ */
+struct stasis_topic *ast_channel_topic_cached(struct ast_channel *chan);
+
+/*!
  * \brief Get the bridge associated with a channel
  * \since 12.0.0
  *
diff --git a/include/asterisk/channel_internal.h b/include/asterisk/channel_internal.h
index a54a1b848..c94cc46f7 100644
--- a/include/asterisk/channel_internal.h
+++ b/include/asterisk/channel_internal.h
@@ -23,5 +23,5 @@ struct ast_channel *__ast_channel_internal_alloc(void (*destructor)(void *obj),
 void ast_channel_internal_finalize(struct ast_channel *chan);
 int ast_channel_internal_is_finalized(struct ast_channel *chan);
 void ast_channel_internal_cleanup(struct ast_channel *chan);
-void ast_channel_internal_setup_topics(struct ast_channel *chan);
+int ast_channel_internal_setup_topics(struct ast_channel *chan);
 
diff --git a/include/asterisk/devicestate.h b/include/asterisk/devicestate.h
index 2b3353ffd..cda8a5d98 100644
--- a/include/asterisk/devicestate.h
+++ b/include/asterisk/devicestate.h
@@ -307,7 +307,14 @@ struct stasis_topic *ast_device_state_topic(const char *device);
  * \retval NULL if it has not been allocated
  * \since 12
  */
-struct stasis_caching_topic *ast_device_state_topic_cached(void);
+struct stasis_topic *ast_device_state_topic_cached(void);
+
+/*!
+ * \brief Backend cache for ast_device_state_topic_cached()
+ * \retval Cache of \ref ast_device_state_message.
+ * \since 12
+ */
+struct stasis_cache *ast_device_state_cache(void);
 
 /*!
  * \brief Get the Stasis message type for device state messages
diff --git a/include/asterisk/presencestate.h b/include/asterisk/presencestate.h
index 5cacf8698..1cdf73eda 100644
--- a/include/asterisk/presencestate.h
+++ b/include/asterisk/presencestate.h
@@ -168,7 +168,14 @@ struct stasis_topic *ast_presence_state_topic_all(void);
  * \retval Caching Stasis topic for presence state messages
  * \since 12
  */
-struct stasis_caching_topic *ast_presence_state_topic_cached(void);
+struct stasis_topic *ast_presence_state_topic_cached(void);
+
+/*!
+ * \brief Backend cache for ast_presence_state_topic_cached()
+ * \retval Cache of \ref ast_presence_state_message.
+ * \since 12
+ */
+struct stasis_cache *ast_presence_state_cache(void);
 
 /*!
  * \brief Stasis message payload representing a presence state update
diff --git a/include/asterisk/stasis.h b/include/asterisk/stasis.h
index fd60724c0..3b9cec34f 100644
--- a/include/asterisk/stasis.h
+++ b/include/asterisk/stasis.h
@@ -94,13 +94,24 @@
  * in the system, and it's desirable to query that state from the cache without
  * locking the original object. It's also desirable for subscribers of the
  * caching topic to receive messages that have both the old cache value and the
- * new value being put into the cache. For this, we have
- * stasis_caching_topic_create(), providing it with the topic which publishes
- * the messages that you wish to cache, and a function that can identify
- * cacheable messages.
- *
- * The returned \ref stasis_caching_topic provides a topic that forwards
- * non-cacheable messages unchanged. A cacheable message is wrapped in a \ref
+ * new value being put into the cache. For this, we have stasis_cache_create()
+ * and stasis_caching_topic_create(), providing them with the topic which
+ * publishes the messages that you wish to cache, and a function that can
+ * identify cacheable messages.
+ *
+ * The \ref stasis_cache is designed so that it may be shared amongst several
+ * \ref stasis_caching_topic objects. This allows you to have individual caching
+ * topics per-object (i.e. so you can subscribe to updates for a single object),
+ * and still have a single cache to query for the state of all objects. While a
+ * cache may be shared amongst different message types, such a usage is probably
+ * not a good idea.
+ *
+ * The \ref stasis_cache can only be written to by \ref stasis_caching_topics.
+ * It's a thread safe container, so freely use the stasis_cache_get() and
+ * stasis_cache_dump() to query the cache.
+ *
+ * The \ref stasis_caching_topic provides a topic that forwards non-cacheable
+ * messages unchanged. A cacheable message is wrapped in a \ref
  * stasis_cache_update message which provides the old snapshot (or \c NULL if
  * this is a new cache entry), and the new snapshot (or \c NULL if the entry was
  * removed from the cache). A stasis_cache_clear_create() message must be sent
@@ -111,6 +122,9 @@
  * stasis_caching_topic will not be freed until after it has been unsubscribed,
  * and all other ao2_ref()'s have been cleaned up.
  *
+ * The \ref stasis_cache object is a normal AO2 managed object, which can be
+ * release with ao2_cleanup().
+ *
  * \par stasis_subscriber
  *
  * Any topic may be subscribed to by simply providing stasis_subscribe() the
@@ -345,6 +359,15 @@ void stasis_forward_message(struct stasis_topic *topic,
 			    struct stasis_topic *publisher_topic,
 			    struct stasis_message *message);
 
+/*!
+ * \brief Wait for all pending messages on a given topic to be processed.
+ * \param topic Topic to await pending messages on.
+ * \return 0 on success.
+ * \return Non-zero on error.
+ * \since 12
+ */
+int stasis_topic_wait(struct stasis_topic *topic);
+
 /*! @} */
 
 /*! @{ */
@@ -514,6 +537,8 @@ struct stasis_message_type *stasis_subscription_change_type(void);
 
 /*! @} */
 
+/*! @{ */
+
 /*!
  * \brief Pool for topic aggregation
  */
@@ -575,22 +600,17 @@ struct stasis_message_type *stasis_cache_clear_type(void);
 /*! @{ */
 
 /*!
- * \brief A topic wrapper, which caches certain messages.
+ * \brief A message cache, for use with \ref stasis_caching_topic.
  * \since 12
  */
-struct stasis_caching_topic;
+struct stasis_cache;
 
 /*!
- * \brief A message which instructs the caching topic to remove an entry from its cache.
- *
- * \param message Message representative of the cache entry that should be cleared.
- *     This will become the data held in the stasis_cache_clear message.
- *
- * \return Message which, when sent to the \a topic, will clear the item from the cache.
- * \return \c NULL on error.
+ * \brief A topic wrapper, which caches certain messages.
  * \since 12
  */
-struct stasis_message *stasis_cache_clear_create(struct stasis_message *message);
+struct stasis_caching_topic;
+
 
 /*!
  * \brief Callback extract a unique identity from a snapshot message.
@@ -606,6 +626,21 @@ struct stasis_message *stasis_cache_clear_create(struct stasis_message *message)
 typedef const char *(*snapshot_get_id)(struct stasis_message *message);
 
 /*!
+ * \brief Create a cache.
+ *
+ * This is the backend store for a \ref stasis_caching_topic. The cache is
+ * thread safe, allowing concurrent reads and writes.
+ *
+ * The returned object is AO2 managed, so ao2_cleanup() when you're done.
+ *
+ * \param id_fn Callback to extract the id from a snapshot message.
+ * \return New cache indexed by \a id_fn.
+ * \return \c NULL on error
+ * \since 12
+ */
+struct stasis_cache *stasis_cache_create(snapshot_get_id id_fn);
+
+/*!
  * \brief Create a topic which monitors and caches messages from another topic.
  *
  * The idea is that some topics publish 'snapshots' of some other object's state
@@ -613,13 +648,17 @@ typedef const char *(*snapshot_get_id)(struct stasis_message *message);
  * is updated, and a stasis_cache_update() message is forwarded, which has both
  * the original snapshot message and the new message.
  *
+ * The returned object is AO2 managed, so ao2_cleanup() when done with it.
+ *
  * \param original_topic Topic publishing snapshot messages.
- * \param id_fn Callback to extract the id from a snapshot message.
+ * \param cache Backend cache in which to keep snapshots.
  * \return New topic which changes snapshot messages to stasis_cache_update()
  *         messages, and forwards all other messages from the original topic.
+ * \return \c NULL on error
  * \since 12
  */
-struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *original_topic, snapshot_get_id id_fn);
+struct stasis_caching_topic *stasis_caching_topic_create(
+	struct stasis_topic *original_topic, struct stasis_cache *cache);
 
 /*!
  * \brief Unsubscribes a caching topic from its upstream topic.
@@ -651,53 +690,55 @@ struct stasis_caching_topic *stasis_caching_unsubscribe_and_join(
 /*!
  * \brief Returns the topic of cached events from a caching topics.
  * \param caching_topic The caching topic.
- * \return The topic that publishes cache update events, along with passthrough events
- *         from the underlying topic.
+ * \return The topic that publishes cache update events, along with passthrough
+ *         events from the underlying topic.
  * \return \c NULL if \a caching_topic is \c NULL.
  * \since 12
  */
-struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *caching_topic);
+struct stasis_topic *stasis_caching_get_topic(
+	struct stasis_caching_topic *caching_topic);
 
 /*!
- * \brief Retrieve an item from the cache.
+ * \brief A message which instructs the caching topic to remove an entry from
+ * its cache.
  *
- * The returned item is AO2 managed, so ao2_cleanup() when you're done with it.
+ * \param message Message representative of the cache entry that should be
+ *                cleared. This will become the data held in the
+ *                stasis_cache_clear message.
  *
- * \param caching_topic The topic returned from stasis_caching_topic_create().
- * \param type Type of message to retrieve.
- * \param id Identity of the snapshot to retrieve.
- * \return Message from the cache.
- * \return \c NULL if message is not found.
+ * \return Message which, when sent to a \ref stasis_caching_topic, will clear
+ *         the item from the cache.
+ * \return \c NULL on error.
  * \since 12
  */
-#define stasis_cache_get(caching_topic, type, id) stasis_cache_get_extended(caching_topic, type, id, 0)
+struct stasis_message *stasis_cache_clear_create(struct stasis_message *message);
 
 /*!
  * \brief Retrieve an item from the cache.
- * \param caching_topic The topic returned from stasis_caching_topic_create().
+ *
+ * The returned item is AO2 managed, so ao2_cleanup() when you're done with it.
+ *
+ * \param cache The cache to query.
  * \param type Type of message to retrieve.
  * \param id Identity of the snapshot to retrieve.
- * \param guaranteed If set to 1 it is guaranteed that any pending messages have been processed.
- * \return Message from the cache. The cache still owns the message, so
- *         ao2_ref() if you want to keep it.
+ * \return Message from the cache.
  * \return \c NULL if message is not found.
  * \since 12
  */
-struct stasis_message *stasis_cache_get_extended(struct stasis_caching_topic *caching_topic,
-					struct stasis_message_type *type,
-					const char *id,
-					unsigned int guaranteed);
+struct stasis_message *stasis_cache_get(
+	struct stasis_cache *cache, struct stasis_message_type *type,
+	const char *id);
 
 /*!
  * \brief Dump cached items to a subscription
- * \param caching_topic The topic returned from stasis_caching_topic_create().
+ * \param cache The cache to query.
  * \param type Type of message to dump (any type if \c NULL).
  * \return ao2_container containing all matches (must be unreffed by caller)
  * \return \c NULL on allocation error
  * \since 12
  */
-struct ao2_container *stasis_cache_dump(struct stasis_caching_topic *caching_topic,
-					struct stasis_message_type *type);
+struct ao2_container *stasis_cache_dump(struct stasis_cache *cache,
+	struct stasis_message_type *type);
 
 /*! @} */
 
@@ -831,13 +872,18 @@ int stasis_cache_init(void);
 
 /*!
  * \internal
- * \brief called by stasis_init for config initialization.
+ * \brief called by stasis_init() for config initialization.
  * \return 0 on success.
  * \return Non-zero on error.
  * \since 12
  */
 int stasis_config_init(void);
 
+/*!
+ * \internal
+ */
+int stasis_wait_init(void);
+
 struct ast_threadpool_options;
 
 /*!
diff --git a/include/asterisk/stasis_bridges.h b/include/asterisk/stasis_bridges.h
index 973b00c47..7fa059fed 100644
--- a/include/asterisk/stasis_bridges.h
+++ b/include/asterisk/stasis_bridges.h
@@ -91,6 +91,22 @@ struct stasis_topic *ast_bridge_topic(struct ast_bridge *bridge);
 
 /*!
  * \since 12
+ * \brief A topic which publishes the events for a particular bridge.
+ *
+ * \ref ast_bridge_snapshot messages are replaced with stasis_cache_update
+ * messages.
+ *
+ * If the given \a bridge is \c NULL, ast_bridge_topic_all_cached() is returned.
+ *
+ * \param bridge Bridge for which to get a topic or \c NULL.
+ *
+ * \retval Topic for bridge's events.
+ * \retval ast_bridge_topic_all() if \a bridge is \c NULL.
+ */
+struct stasis_topic *ast_bridge_topic_cached(struct ast_bridge *bridge);
+
+/*!
+ * \since 12
  * \brief A topic which publishes the events for all bridges.
  * \retval Topic for all bridge events.
  */
@@ -103,7 +119,14 @@ struct stasis_topic *ast_bridge_topic_all(void);
  *
  * \retval Caching topic for all bridge events.
  */
-struct stasis_caching_topic *ast_bridge_topic_all_cached(void);
+struct stasis_topic *ast_bridge_topic_all_cached(void);
+
+/*!
+ * \since 12
+ * \brief Backend cache for ast_bridge_topic_all_cached().
+ * \retval Cache of \ref ast_bridge_snapshot.
+ */
+struct stasis_cache *ast_bridge_cache(void);
 
 /*!
  * \since 12
@@ -408,6 +431,15 @@ struct ast_bridge_snapshot *ast_bridge_snapshot_get_latest(
 	const char *bridge_id);
 
 /*!
+ * \internal
+ * \brief Initialize the topics for a single bridge.
+ * \return 0 on success.
+ * \return Non-zero on error.
+ */
+int bridge_topics_init(struct ast_bridge *bridge);
+
+/*!
+ * \internal
  * \brief Initialize the stasis bridging topic and message types
  * \retval 0 on success
  * \retval -1 on failure
diff --git a/include/asterisk/stasis_cache_pattern.h b/include/asterisk/stasis_cache_pattern.h
new file mode 100644
index 000000000..2ea643e19
--- /dev/null
+++ b/include/asterisk/stasis_cache_pattern.h
@@ -0,0 +1,153 @@
+/*
+ * Asterisk -- An open source telephony toolkit.
+ *
+ * Copyright (C) 2013, Digium, Inc.
+ *
+ * David M. Lee, II <dlee@digium.com>
+ *
+ * See http://www.asterisk.org for more information about
+ * the Asterisk project. Please do not directly contact
+ * any of the maintainers of this project for assistance;
+ * the project provides a web site, mailing lists and IRC
+ * channels for your use.
+ *
+ * This program is free software, distributed under the terms of
+ * the GNU General Public License Version 2. See the LICENSE file
+ * at the top of the source tree.
+ */
+
+#ifndef _ASTERISK_STASIS_CACHE_PATTERN_H
+#define _ASTERISK_STASIS_CACHE_PATTERN_H
+
+/*! \file
+ *
+ * \brief Caching pattern for \ref stasis topics.
+ *
+ * A typical pattern for Stasis objects is to have individual objects, which
+ * have their own topic and caching topic. These individual topics feed an
+ * upstream aggregate topics, and a shared cache.
+ *
+ * The \ref stasis_cp_all object contains the aggregate topics and shared cache.
+ * This is built with the base name for the topics, and the identity function to
+ * identify messages in the cache.
+ *
+ * The \ref stasis_cp_single object contains the \ref stasis_topic for a single
+ * instance, and the corresponding \ref stasis_caching_topic.
+ *
+ * Since the \ref stasis_cp_single object has subscriptions for forwarding
+ * and caching, it must be disposed of using stasis_cp_single_unsubscribe()
+ * instead of simply ao2_cleanup().
+ */
+
+#include "asterisk/stasis.h"
+
+/*!
+ * \brief The 'all' side of the cache pattern. These are typically built as
+ * global objects for specific modules.
+ */
+struct stasis_cp_all;
+
+/*!
+ * \brief Create an all instance of the cache pattern.
+ *
+ * This object is AO2 managed, so dispose of it with ao2_cleanup().
+ *
+ * \param name Base name of the topics.
+ * \param id_fn Identity function for the cache.
+ * \return All side instance.
+ * \return \c NULL on error.
+ */
+struct stasis_cp_all *stasis_cp_all_create(const char *name,
+	snapshot_get_id id_fn);
+
+/*!
+ * \brief Get the aggregate topic.
+ *
+ * This topic aggregates all messages published to corresponding
+ * stasis_cp_single_topic() topics.
+ *
+ * \param all All side caching pattern object.
+ * \return The aggregate topic.
+ * \return \c NULL if \a all is \c NULL
+ */
+struct stasis_topic *stasis_cp_all_topic(struct stasis_cp_all *all);
+
+/*!
+ * \brief Get the caching topic.
+ *
+ * This topic aggregates all messages from the corresponding
+ * stasis_cp_single_topic_cached() topics.
+ *
+ * Note that one normally only subscribes to the caching topic, since data
+ * is fed to it from its upstream topic.
+ *
+ * \param all All side caching pattern object.
+ * \return The aggregate caching topic.
+ * \return \c NULL if \a all is \c NULL
+ */
+struct stasis_topic *stasis_cp_all_topic_cached(
+	struct stasis_cp_all *all);
+
+/*!
+ * \brief Get the cache.
+ *
+ * This is the shared cache for all corresponding \ref stasis_cp_single objects.
+ *
+ * \param all All side caching pattern object.
+ * \return The cache.
+ * \return \c NULL if \a all is \c NULL
+ */
+struct stasis_cache *stasis_cp_all_cache(struct stasis_cp_all *all);
+
+/*!
+ * \brief The 'one' side of the cache pattern. These are built per-instance for
+ * some corresponding object, and must be explicitly disposed of using
+ * stasis_cp_single_unsubscribe().
+ */
+struct stasis_cp_single;
+
+/*!
+ * \brief Create the 'one' side of the cache pattern.
+ *
+ * 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(struct stasis_cp_all *all,
+	const char *name);
+
+/*!
+ * \brief Stops caching and forwarding messages.
+ *
+ * \param one One side of the cache pattern.
+ */
+void stasis_cp_single_unsubscribe(struct stasis_cp_single *one);
+
+/*!
+ * \brief Get the topic for this instance.
+ *
+ * This is the topic to which one would post instance-specific messages, or
+ * subscribe for single-instance, uncached messages.
+ *
+ * \param one One side of the cache pattern.
+ * \return The main topic.
+ * \return \c NULL if \a one is \c NULL
+ */
+struct stasis_topic *stasis_cp_single_topic(struct stasis_cp_single *one);
+
+/*!
+ * \brief Get the caching topic for this instance.
+ *
+ * Note that one normally only subscribes to the caching topic, since data
+ * is fed to it from its upstream topic.
+ *
+ * \param one One side of the cache pattern.
+ * \return The caching topic.
+ * \return \c NULL if \a one is \c NULL
+ */
+struct stasis_topic *stasis_cp_single_topic_cached(
+	struct stasis_cp_single *one);
+
+#endif /* _ASTERISK_STASIS_CACHE_PATTERN_H */
diff --git a/include/asterisk/stasis_channels.h b/include/asterisk/stasis_channels.h
index 8cc4c8382..c0b596ac9 100644
--- a/include/asterisk/stasis_channels.h
+++ b/include/asterisk/stasis_channels.h
@@ -106,6 +106,8 @@ struct ast_channel_blob {
  */
 struct ast_multi_channel_blob;
 
+struct stasis_cp_all *ast_channel_cache_all(void);
+
 /*!
  * \since 12
  * \brief A topic which publishes the events for all channels.
@@ -120,16 +122,23 @@ struct stasis_topic *ast_channel_topic_all(void);
  *
  * \retval Topic for all channel events.
  */
-struct stasis_caching_topic *ast_channel_topic_all_cached(void);
+struct stasis_topic *ast_channel_topic_all_cached(void);
 
 /*!
  * \since 12
- * \brief A caching topic which caches \ref ast_channel_snapshot messages from
- * ast_channel_events_all(void) and indexes them by name.
+ * \brief Primary channel cache, indexed by Uniqueid.
  *
- * \retval Topic for all channel events.
+ * \retval Cache of \ref ast_channel_snapshot.
+ */
+struct stasis_cache *ast_channel_cache(void);
+
+/*!
+ * \since 12
+ * \brief Secondary channel cache, indexed by name.
+ *
+ * \retval Cache of \ref ast_channel_snapshot.
  */
-struct stasis_caching_topic *ast_channel_topic_all_cached_by_name(void);
+struct stasis_cache *ast_channel_cache_by_name(void);
 
 /*!
  * \since 12
@@ -551,7 +560,9 @@ int ast_channel_snapshot_caller_id_equal(
 
 /*!
  * \brief Initialize the stasis channel topic and message types
+ * \return 0 on success
+ * \return Non-zero on error
  */
-void ast_stasis_channels_init(void);
+int ast_stasis_channels_init(void);
 
 #endif /* STASIS_CHANNELS_H_ */
diff --git a/include/asterisk/stasis_endpoints.h b/include/asterisk/stasis_endpoints.h
index 6d7f8dbaa..4a35e9587 100644
--- a/include/asterisk/stasis_endpoints.h
+++ b/include/asterisk/stasis_endpoints.h
@@ -30,6 +30,7 @@
 #include "asterisk/endpoints.h"
 #include "asterisk/json.h"
 #include "asterisk/stasis.h"
+#include "asterisk/stasis_cache_pattern.h"
 #include "asterisk/stringfields.h"
 
 /*! \addtogroup StasisTopicsAndMessages
@@ -144,6 +145,31 @@ struct ast_endpoint_snapshot *ast_endpoint_snapshot_create(
 struct stasis_topic *ast_endpoint_topic(struct ast_endpoint *endpoint);
 
 /*!
+ * \brief Returns the topic for a specific endpoint.
+ *
+ * \ref ast_endpoint_snapshot messages are replaced with
+ * \ref stasis_cache_update
+ *
+ * \param endpoint The endpoint.
+ * \return The topic for the given endpoint.
+ * \return ast_endpoint_topic_all() if endpoint is \c NULL.
+ * \since 12
+ */
+struct stasis_topic *ast_endpoint_topic_cached(struct ast_endpoint *endpoint);
+
+/*!
+ * \internal
+ * \brief Cache and global topics for endpoints.
+ *
+ * This is public simply to be used by endpoints.c. Please use the accessor
+ * functions (ast_endpoint_topic_all(), ast_endpoint_topic_all_cached(),
+ * ast_endpoint_cache(), etc.) instead of calling this directly.
+ *
+ * \since 12
+ */
+struct stasis_cp_all *ast_endpoint_cache_all(void);
+
+/*!
  * \brief Topic for all endpoint releated messages.
  * \since 12
  */
@@ -153,7 +179,14 @@ struct stasis_topic *ast_endpoint_topic_all(void);
  * \brief Cached topic for all endpoint related messages.
  * \since 12
  */
-struct stasis_caching_topic *ast_endpoint_topic_all_cached(void);
+struct stasis_topic *ast_endpoint_topic_all_cached(void);
+
+/*!
+ * \brief Backend cache for ast_endpoint_topic_all_cached().
+ * \return Cache of \ref ast_endpoint_snapshot.
+ * \since 12
+ */
+struct stasis_cache *ast_endpoint_cache(void);
 
 /*!
  * \brief Retrieve the most recent snapshot for the endpoint with the given