[asterisk-commits] file: trunk r391596 - in /trunk: include/asterisk/ main/ res/stasis_http/
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Thu Jun 13 06:02:22 CDT 2013
Author: file
Date: Thu Jun 13 06:02:16 2013
New Revision: 391596
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=391596
Log:
Add support for requiring that all queued messages on a caching topic have been handled before
retrieving from the cache and also change adding channels to an endpoint to be an immediate
operation.
Review: https://reviewboard.asterisk.org/r/2599/
Modified:
trunk/include/asterisk/channel.h
trunk/include/asterisk/stasis.h
trunk/include/asterisk/stasis_endpoints.h
trunk/main/channel_internal_api.c
trunk/main/endpoints.c
trunk/main/stasis_cache.c
trunk/main/stasis_endpoints.c
trunk/res/stasis_http/resource_endpoints.c
Modified: trunk/include/asterisk/channel.h
URL: http://svnview.digium.com/svn/asterisk/trunk/include/asterisk/channel.h?view=diff&rev=391596&r1=391595&r2=391596
==============================================================================
--- trunk/include/asterisk/channel.h (original)
+++ trunk/include/asterisk/channel.h Thu Jun 13 06:02:16 2013
@@ -152,6 +152,7 @@
#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 @@
*/
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 */
Modified: trunk/include/asterisk/stasis.h
URL: http://svnview.digium.com/svn/asterisk/trunk/include/asterisk/stasis.h?view=diff&rev=391596&r1=391595&r2=391596
==============================================================================
--- trunk/include/asterisk/stasis.h (original)
+++ trunk/include/asterisk/stasis.h Thu Jun 13 06:02:16 2013
@@ -670,9 +670,23 @@
* \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
Modified: trunk/include/asterisk/stasis_endpoints.h
URL: http://svnview.digium.com/svn/asterisk/trunk/include/asterisk/stasis_endpoints.h?view=diff&rev=391596&r1=391595&r2=391596
==============================================================================
--- trunk/include/asterisk/stasis_endpoints.h (original)
+++ trunk/include/asterisk/stasis_endpoints.h Thu Jun 13 06:02:16 2013
@@ -122,12 +122,14 @@
*
* \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
);
/*! @} */
Modified: trunk/main/channel_internal_api.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/channel_internal_api.c?view=diff&rev=391596&r1=391595&r2=391596
==============================================================================
--- trunk/main/channel_internal_api.c (original)
+++ trunk/main/channel_internal_api.c Thu Jun 13 06:02:16 2013
@@ -1406,8 +1406,7 @@
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);
@@ -1418,8 +1417,6 @@
if (chan->endpoint_forward == NULL) {
return -1;
}
-
- ast_publish_channel_state(chan);
return 0;
}
Modified: trunk/main/endpoints.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/endpoints.c?view=diff&rev=391596&r1=391595&r2=391596
==============================================================================
--- trunk/main/endpoints.c (original)
+++ trunk/main/endpoints.c Thu Jun 13 06:02:16 2013
@@ -122,30 +122,23 @@
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;
-
- ast_assert(endpoint != NULL);
- ast_assert(snapshot != NULL);
+
+int ast_endpoint_add_channel(struct ast_endpoint *endpoint,
+ struct ast_channel *chan)
+{
+ ast_assert(chan != NULL);
+ ast_assert(endpoint != 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 */
@@ -236,9 +229,6 @@
if (!endpoint->router) {
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);
Modified: trunk/main/stasis_cache.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/stasis_cache.c?view=diff&rev=391596&r1=391595&r2=391596
==============================================================================
--- trunk/main/stasis_cache.c (original)
+++ trunk/main/stasis_cache.c Thu Jun 13 06:02:16 2013
@@ -46,9 +46,12 @@
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;
@@ -60,6 +63,8 @@
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,12 +209,61 @@
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) {
@@ -261,6 +315,7 @@
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 @@
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 @@
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 @@
{
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 @@
return -1;
}
+ if (STASIS_MESSAGE_TYPE_INIT(cache_guarantee_type) != 0) {
+ return -1;
+ }
+
return 0;
}
Modified: trunk/main/stasis_endpoints.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/stasis_endpoints.c?view=diff&rev=391596&r1=391595&r2=391596
==============================================================================
--- trunk/main/stasis_endpoints.c (original)
+++ trunk/main/stasis_endpoints.c Thu Jun 13 06:02:16 2013
@@ -52,7 +52,7 @@
}
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 @@
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;
}
Modified: trunk/res/stasis_http/resource_endpoints.c
URL: http://svnview.digium.com/svn/asterisk/trunk/res/stasis_http/resource_endpoints.c?view=diff&rev=391596&r1=391595&r2=391596
==============================================================================
--- trunk/res/stasis_http/resource_endpoints.c (original)
+++ trunk/res/stasis_http/resource_endpoints.c Thu Jun 13 06:02:16 2013
@@ -140,7 +140,7 @@
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");
More information about the asterisk-commits
mailing list