Add support for requiring that all queued messages on a caching topic have been handl...
authorJoshua Colp <jcolp@digium.com>
Thu, 13 Jun 2013 11:02:16 +0000 (11:02 +0000)
committerJoshua Colp <jcolp@digium.com>
Thu, 13 Jun 2013 11:02:16 +0000 (11:02 +0000)
retrieving from the cache and also change adding channels to an endpoint to be an immediate
operation.

Review: https://reviewboard.asterisk.org/r/2599/

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

include/asterisk/channel.h
include/asterisk/stasis.h
include/asterisk/stasis_endpoints.h
main/channel_internal_api.c
main/endpoints.c
main/stasis_cache.c
main/stasis_endpoints.c
res/stasis_http/resource_endpoints.c

index c2edc18..42d50f2 100644 (file)
@@ -152,6 +152,7 @@ extern "C" {
 #include "asterisk/framehook.h"
 #include "asterisk/stasis.h"
 #include "asterisk/json.h"
+#include "asterisk/endpoints.h"
 
 #define DATASTORE_INHERIT_FOREVER      INT_MAX
 
@@ -4265,4 +4266,16 @@ struct ast_channel *ast_channel_yank(struct ast_channel *yankee);
  */
 int ast_channel_move(struct ast_channel *dest, struct ast_channel *source);
 
+/*!
+ * \since 12
+ * \brief Forward channel stasis messages to the given endpoint
+ *
+ * \param chan The channel to forward from
+ * \param endpoint The endpoint to forward to
+ *
+ * \retval 0 Success
+ * \retval non-zero Failure
+ */
+int ast_channel_forward_endpoint(struct ast_channel *chan, struct ast_endpoint *endpoint);
+
 #endif /* _ASTERISK_CHANNEL_H */
index bd303eb..19ebb41 100644 (file)
@@ -670,9 +670,23 @@ struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *cachi
  * \return \c NULL if message is not found.
  * \since 12
  */
-struct stasis_message *stasis_cache_get(struct stasis_caching_topic *caching_topic,
+#define stasis_cache_get(caching_topic, type, id) stasis_cache_get_extended(caching_topic, type, id, 0)
+
+/*!
+ * \brief Retrieve an item from the cache.
+ * \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.
+ * \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 \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);
+                                       const char *id,
+                                       unsigned int guaranteed);
 
 /*!
  * \brief Dump cached items to a subscription
index 2aeb614..10abcd7 100644 (file)
@@ -122,12 +122,14 @@ struct stasis_caching_topic *ast_endpoint_topic_all_cached(void);
  *
  * \param tech Name of the endpoint's technology.
  * \param resource Resource name of the endpoint.
+ * \param guaranteed Whether to require all pending messages to have been processed or not.
  * \return Snapshot of the endpoint with the given name.
  * \return \c NULL if endpoint is not found, or on error.
  * \since 12
  */
 struct ast_endpoint_snapshot *ast_endpoint_latest_snapshot(const char *tech,
-       const char *resource
+       const char *resource,
+       unsigned int guaranteed
 );
 
 /*! @} */
index 4e5284c..1e21cce 100644 (file)
@@ -1406,8 +1406,7 @@ struct stasis_topic *ast_channel_topic(struct ast_channel *chan)
        return chan ? chan->topic : ast_channel_topic_all();
 }
 
-int ast_endpoint_add_channel(struct ast_endpoint *endpoint,
-       struct ast_channel *chan)
+int ast_channel_forward_endpoint(struct ast_channel *chan, struct ast_endpoint *endpoint)
 {
        ast_assert(chan != NULL);
        ast_assert(endpoint != NULL);
@@ -1419,8 +1418,6 @@ int ast_endpoint_add_channel(struct ast_endpoint *endpoint,
                return -1;
        }
 
-       ast_publish_channel_state(chan);
-
        return 0;
 }
 
index 9f6d2c0..d689f2e 100644 (file)
@@ -122,30 +122,23 @@ static void endpoint_dtor(void *obj)
        ast_string_field_free_memory(endpoint);
 }
 
-static void endpoint_channel_snapshot(void *data,
-       struct stasis_subscription *sub, struct stasis_topic *topic,
-       struct stasis_message *message)
-{
-       struct ast_endpoint *endpoint = data;
-       struct ast_channel_snapshot *snapshot = stasis_message_data(message);
-       RAII_VAR(char *, existing_id, NULL, ao2_cleanup);
-       int publish = 0;
 
+int ast_endpoint_add_channel(struct ast_endpoint *endpoint,
+       struct ast_channel *chan)
+{
+       ast_assert(chan != NULL);
        ast_assert(endpoint != NULL);
-       ast_assert(snapshot != NULL);
+
+       ast_channel_forward_endpoint(chan, endpoint);
 
        ao2_lock(endpoint);
-       existing_id = ao2_find(endpoint->channel_ids, snapshot->uniqueid,
-               OBJ_POINTER);
-       if (!existing_id) {
-               ast_str_container_add(endpoint->channel_ids,
-                       snapshot->uniqueid);
-               publish = 1;
-       }
+       ast_str_container_add(endpoint->channel_ids, ast_channel_uniqueid(chan));
        ao2_unlock(endpoint);
-       if (publish) {
-               endpoint_publish_snapshot(endpoint);
-       }
+
+       ast_publish_channel_state(chan);
+       endpoint_publish_snapshot(endpoint);
+
+       return 0;
 }
 
 /*! \brief Handler for channel snapshot cache clears */
@@ -237,9 +230,6 @@ struct ast_endpoint *ast_endpoint_create(const char *tech, const char *resource)
                return NULL;
        }
        r |= stasis_message_router_add(endpoint->router,
-               ast_channel_snapshot_type(), endpoint_channel_snapshot,
-               endpoint);
-       r |= stasis_message_router_add(endpoint->router,
                stasis_cache_clear_type(), endpoint_cache_clear,
                endpoint);
        r |= stasis_message_router_set_default(endpoint->router,
index 546ad49..115bb7b 100644 (file)
@@ -46,10 +46,13 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
 struct stasis_caching_topic {
        struct ao2_container *cache;
        struct stasis_topic *topic;
+       struct stasis_topic *original_topic;
        struct stasis_subscription *sub;
        snapshot_get_id id_fn;
 };
 
+static struct stasis_message_type *cache_guarantee_type(void);
+
 static void stasis_caching_topic_dtor(void *obj) {
        struct stasis_caching_topic *caching_topic = obj;
        ast_assert(!stasis_subscription_is_subscribed(caching_topic->sub));
@@ -60,6 +63,8 @@ static void stasis_caching_topic_dtor(void *obj) {
        caching_topic->cache = NULL;
        ao2_cleanup(caching_topic->topic);
        caching_topic->topic = NULL;
+       ao2_cleanup(caching_topic->original_topic);
+       caching_topic->original_topic = NULL;
 }
 
 struct stasis_topic *stasis_caching_get_topic(struct stasis_caching_topic *caching_topic)
@@ -204,13 +209,62 @@ static struct stasis_message *cache_put(struct stasis_caching_topic *caching_top
        return old_snapshot;
 }
 
-struct stasis_message *stasis_cache_get(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, const char *id)
+/*! \internal */
+struct caching_guarantee {
+       ast_mutex_t lock;
+       ast_cond_t cond;
+       unsigned int done:1;
+};
+
+static void caching_guarantee_dtor(void *obj)
+{
+       struct caching_guarantee *guarantee = obj;
+
+       ast_assert(guarantee->done == 1);
+
+       ast_mutex_destroy(&guarantee->lock);
+       ast_cond_destroy(&guarantee->cond);
+}
+
+static struct stasis_message *caching_guarantee_create(void)
+{
+       RAII_VAR(struct caching_guarantee *, guarantee, NULL, ao2_cleanup);
+       RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+
+       if (!(guarantee = ao2_alloc(sizeof(*guarantee), caching_guarantee_dtor))) {
+               return NULL;
+       }
+
+       ast_mutex_init(&guarantee->lock);
+       ast_cond_init(&guarantee->cond, NULL);
+
+       if (!(msg = stasis_message_create(cache_guarantee_type(), guarantee))) {
+               return NULL;
+       }
+
+       ao2_ref(msg, +1);
+       return msg;
+}
+
+struct stasis_message *stasis_cache_get_extended(struct stasis_caching_topic *caching_topic, struct stasis_message_type *type, const char *id, unsigned int guaranteed)
 {
        RAII_VAR(struct cache_entry *, search_entry, NULL, ao2_cleanup);
        RAII_VAR(struct cache_entry *, cached_entry, NULL, ao2_cleanup);
 
        ast_assert(caching_topic->cache != NULL);
 
+       if (guaranteed) {
+               RAII_VAR(struct stasis_message *, msg, caching_guarantee_create(), ao2_cleanup);
+               struct caching_guarantee *guarantee = stasis_message_data(msg);
+
+               ast_mutex_lock(&guarantee->lock);
+               stasis_publish(caching_topic->original_topic, msg);
+               while (!guarantee->done) {
+                       ast_cond_wait(&guarantee->cond, &guarantee->lock);
+               }
+               ast_mutex_unlock(&guarantee->lock);
+       }
+
        search_entry = cache_entry_create(type, id, NULL);
        if (search_entry == NULL) {
                return NULL;
@@ -261,6 +315,7 @@ struct ao2_container *stasis_cache_dump(struct stasis_caching_topic *caching_top
 
 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type);
 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_update_type);
+STASIS_MESSAGE_TYPE_DEFN(cache_guarantee_type);
 
 struct stasis_message *stasis_cache_clear_create(struct stasis_message *id_message)
 {
@@ -340,6 +395,18 @@ static void caching_topic_exec(void *data, struct stasis_subscription *sub, stru
                caching_topic_needs_unref = caching_topic;
        }
 
+       /* Handle cache guarantee event */
+       if (cache_guarantee_type() == stasis_message_type(message)) {
+               struct caching_guarantee *guarantee = stasis_message_data(message);
+
+               ast_mutex_lock(&guarantee->lock);
+               guarantee->done = 1;
+               ast_cond_signal(&guarantee->cond);
+               ast_mutex_unlock(&guarantee->lock);
+
+               return;
+       }
+
        /* Handle cache clear event */
        if (stasis_cache_clear_type() == stasis_message_type(message)) {
                RAII_VAR(struct stasis_message *, old_snapshot, NULL, ao2_cleanup);
@@ -423,6 +490,10 @@ struct stasis_caching_topic *stasis_caching_topic_create(struct stasis_topic *or
        if (sub == NULL) {
                return NULL;
        }
+
+       ao2_ref(original_topic, +1);
+       caching_topic->original_topic = original_topic;
+
        /* This is for the reference contained in the subscription above */
        ao2_ref(caching_topic, +1);
        caching_topic->sub = sub;
@@ -435,6 +506,7 @@ static void stasis_cache_cleanup(void)
 {
        STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_clear_type);
        STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_update_type);
+       STASIS_MESSAGE_TYPE_CLEANUP(cache_guarantee_type);
 }
 
 int stasis_cache_init(void)
@@ -449,6 +521,10 @@ int stasis_cache_init(void)
                return -1;
        }
 
+       if (STASIS_MESSAGE_TYPE_INIT(cache_guarantee_type) != 0) {
+               return -1;
+       }
+
        return 0;
 }
 
index a7b7f55..d5347cb 100644 (file)
@@ -52,7 +52,7 @@ struct stasis_caching_topic *ast_endpoint_topic_all_cached(void)
 }
 
 struct ast_endpoint_snapshot *ast_endpoint_latest_snapshot(const char *tech,
-       const char *name)
+       const char *name, unsigned int guaranteed)
 {
        RAII_VAR(char *, id, NULL, ast_free);
        RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
@@ -63,8 +63,8 @@ struct ast_endpoint_snapshot *ast_endpoint_latest_snapshot(const char *tech,
                return NULL;
        }
 
-       msg = stasis_cache_get(ast_endpoint_topic_all_cached(),
-               ast_endpoint_snapshot_type(), id);
+       msg = stasis_cache_get_extended(ast_endpoint_topic_all_cached(),
+               ast_endpoint_snapshot_type(), id, guaranteed);
        if (!msg) {
                return NULL;
        }
index 52d05c0..306413d 100644 (file)
@@ -140,7 +140,7 @@ void stasis_http_get_endpoint(struct ast_variable *headers,
        RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
        RAII_VAR(struct ast_endpoint_snapshot *, snapshot, NULL, ao2_cleanup);
 
-       snapshot = ast_endpoint_latest_snapshot(args->tech, args->resource);
+       snapshot = ast_endpoint_latest_snapshot(args->tech, args->resource, 0);
        if (!snapshot) {
                stasis_http_response_error(response, 404, "Not Found",
                        "Endpoint not found");