[asterisk-commits] file: branch file/stasis_tweaking r390501 - in /team/file/stasis_tweaking: in...

SVN commits to the Asterisk project asterisk-commits at lists.digium.com
Wed Jun 5 10:53:33 CDT 2013


Author: file
Date: Wed Jun  5 10:53:31 2013
New Revision: 390501

URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=390501
Log:
Add the ability to retrieve a message from a stasis cache with the guarantee that all pending changes have been applied.

Modified:
    team/file/stasis_tweaking/include/asterisk/stasis.h
    team/file/stasis_tweaking/include/asterisk/stasis_endpoints.h
    team/file/stasis_tweaking/main/stasis_cache.c
    team/file/stasis_tweaking/main/stasis_endpoints.c
    team/file/stasis_tweaking/res/stasis_http/resource_endpoints.c

Modified: team/file/stasis_tweaking/include/asterisk/stasis.h
URL: http://svnview.digium.com/svn/asterisk/team/file/stasis_tweaking/include/asterisk/stasis.h?view=diff&rev=390501&r1=390500&r2=390501
==============================================================================
--- team/file/stasis_tweaking/include/asterisk/stasis.h (original)
+++ team/file/stasis_tweaking/include/asterisk/stasis.h Wed Jun  5 10:53:31 2013
@@ -613,9 +613,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 snapshot changes have been applied.
+ * \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: team/file/stasis_tweaking/include/asterisk/stasis_endpoints.h
URL: http://svnview.digium.com/svn/asterisk/team/file/stasis_tweaking/include/asterisk/stasis_endpoints.h?view=diff&rev=390501&r1=390500&r2=390501
==============================================================================
--- team/file/stasis_tweaking/include/asterisk/stasis_endpoints.h (original)
+++ team/file/stasis_tweaking/include/asterisk/stasis_endpoints.h Wed Jun  5 10:53:31 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 changes to have been applied 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: team/file/stasis_tweaking/main/stasis_cache.c
URL: http://svnview.digium.com/svn/asterisk/team/file/stasis_tweaking/main/stasis_cache.c?view=diff&rev=390501&r1=390500&r2=390501
==============================================================================
--- team/file/stasis_tweaking/main/stasis_cache.c (original)
+++ team/file/stasis_tweaking/main/stasis_cache.c Wed Jun  5 10:53:31 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 *stasis_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,59 @@
 	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 stasis_caching_guarantee {
+	ast_mutex_t lock;
+	ast_cond_t cond;
+	unsigned int done:1;
+};
+
+static void stasis_caching_guarantee_dtor(void *obj)
+{
+	struct stasis_caching_guarantee *guarantee = obj;
+
+	ast_mutex_destroy(&guarantee->lock);
+	ast_cond_destroy(&guarantee->cond);
+}
+
+static struct stasis_message *stasis_caching_guarantee_create(void)
+{
+	RAII_VAR(struct stasis_caching_guarantee *, guarantee, NULL, ao2_cleanup);
+	RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+
+	if (!(guarantee = ao2_alloc(sizeof(*guarantee), stasis_caching_guarantee_dtor))) {
+		return NULL;
+	}
+
+	ast_mutex_init(&guarantee->lock);
+	ast_cond_init(&guarantee->cond, NULL);
+
+	if (!(msg = stasis_message_create(stasis_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, stasis_caching_guarantee_create(), ao2_cleanup);
+		struct stasis_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 +313,7 @@
 
 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_clear_type);
 STASIS_MESSAGE_TYPE_DEFN(stasis_cache_update_type);
+STASIS_MESSAGE_TYPE_DEFN(stasis_cache_guarantee_type);
 
 static void cache_clear_dtor(void *obj)
 {
@@ -357,6 +410,18 @@
 
 	if (stasis_subscription_final_message(sub, message)) {
 		caching_topic_needs_unref = caching_topic;
+	}
+
+	/* Handle cache guarantee event */
+	if (stasis_cache_guarantee_type() == stasis_message_type(message)) {
+		struct stasis_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 */
@@ -438,6 +503,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;
@@ -450,6 +519,7 @@
 {
 	STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_clear_type);
 	STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_update_type);
+	STASIS_MESSAGE_TYPE_CLEANUP(stasis_cache_guarantee_type);
 }
 
 int stasis_cache_init(void)
@@ -464,6 +534,10 @@
 		return -1;
 	}
 
+	if (STASIS_MESSAGE_TYPE_INIT(stasis_cache_guarantee_type) != 0) {
+		return -1;
+	}
+
 	return 0;
 }
 

Modified: team/file/stasis_tweaking/main/stasis_endpoints.c
URL: http://svnview.digium.com/svn/asterisk/team/file/stasis_tweaking/main/stasis_endpoints.c?view=diff&rev=390501&r1=390500&r2=390501
==============================================================================
--- team/file/stasis_tweaking/main/stasis_endpoints.c (original)
+++ team/file/stasis_tweaking/main/stasis_endpoints.c Wed Jun  5 10:53:31 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: team/file/stasis_tweaking/res/stasis_http/resource_endpoints.c
URL: http://svnview.digium.com/svn/asterisk/team/file/stasis_tweaking/res/stasis_http/resource_endpoints.c?view=diff&rev=390501&r1=390500&r2=390501
==============================================================================
--- team/file/stasis_tweaking/res/stasis_http/resource_endpoints.c (original)
+++ team/file/stasis_tweaking/res/stasis_http/resource_endpoints.c Wed Jun  5 10:53:31 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