[asterisk-commits] kmoore: branch kmoore/stasis-mwi r382372 - /team/kmoore/stasis-mwi/apps/

SVN commits to the Asterisk project asterisk-commits at lists.digium.com
Mon Mar 4 09:39:28 CST 2013


Author: kmoore
Date: Mon Mar  4 09:39:24 2013
New Revision: 382372

URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=382372
Log:
Transition pollmailboxes to Stasis

Modified:
    team/kmoore/stasis-mwi/apps/app_voicemail.c

Modified: team/kmoore/stasis-mwi/apps/app_voicemail.c
URL: http://svnview.digium.com/svn/asterisk/team/kmoore/stasis-mwi/apps/app_voicemail.c?view=diff&rev=382372&r1=382371&r2=382372
==============================================================================
--- team/kmoore/stasis-mwi/apps/app_voicemail.c (original)
+++ team/kmoore/stasis-mwi/apps/app_voicemail.c Mon Mar  4 09:39:24 2013
@@ -974,10 +974,8 @@
 static pthread_t poll_thread = AST_PTHREADT_NULL;
 static unsigned char poll_thread_run;
 
-/*! Subscription to ... MWI event subscriptions */
-static struct ast_event_sub *mwi_sub_sub;
-/*! Subscription to ... MWI event un-subscriptions */
-static struct ast_event_sub *mwi_unsub_sub;
+/*! Subscription to MWI event subscription changes */
+static struct stasis_subscription *mwi_sub_sub;
 
 /*!
  * \brief An MWI subscription
@@ -991,14 +989,14 @@
 	int old_urgent;
 	int old_new;
 	int old_old;
-	uint32_t uniqueid;
+	char *uniqueid;
 	char mailbox[1];
 };
 
 struct mwi_sub_task {
 	const char *mailbox;
 	const char *context;
-	uint32_t uniqueid;
+	const char *uniqueid;
 };
 
 static struct ast_taskprocessor *mwi_subscription_tps;
@@ -12534,26 +12532,26 @@
 
 static void mwi_sub_destroy(struct mwi_sub *mwi_sub)
 {
+	ast_free(mwi_sub->uniqueid);
 	ast_free(mwi_sub);
 }
 
 static int handle_unsubscribe(void *datap)
 {
 	struct mwi_sub *mwi_sub;
-	uint32_t *uniqueid = datap;
+	char *uniqueid = datap;
 	
 	AST_RWLIST_WRLOCK(&mwi_subs);
 	AST_RWLIST_TRAVERSE_SAFE_BEGIN(&mwi_subs, mwi_sub, entry) {
-		if (mwi_sub->uniqueid == *uniqueid) {
+		if (!strcmp(mwi_sub->uniqueid, uniqueid)) {
 			AST_LIST_REMOVE_CURRENT(entry);
-			break;
+			/* Don't break here since a duplicate uniqueid
+			 * may have been added as a result of a cache dump. */
+			mwi_sub_destroy(mwi_sub);
 		}
 	}
 	AST_RWLIST_TRAVERSE_SAFE_END
 	AST_RWLIST_UNLOCK(&mwi_subs);
-
-	if (mwi_sub)
-		mwi_sub_destroy(mwi_sub);
 
 	ast_free(uniqueid);	
 	return 0;
@@ -12575,7 +12573,7 @@
 	if (!(mwi_sub = ast_calloc(1, len)))
 		return -1;
 
-	mwi_sub->uniqueid = p->uniqueid;
+	mwi_sub->uniqueid = ast_strdup(p->uniqueid);
 	if (!ast_strlen_zero(p->mailbox))
 		strcpy(mwi_sub->mailbox, p->mailbox);
 
@@ -12589,73 +12587,83 @@
 	AST_RWLIST_UNLOCK(&mwi_subs);
 	ast_free((void *) p->mailbox);
 	ast_free((void *) p->context);
+	ast_free((void *) p->uniqueid);
 	ast_free(p);
 	poll_subscribed_mailbox(mwi_sub);
 	return 0;
 }
 
-static void mwi_unsub_event_cb(const struct ast_event *event, void *userdata)
-{
-	uint32_t u, *uniqueid = ast_calloc(1, sizeof(*uniqueid));
+static void mwi_unsub_event_cb(struct stasis_subscription_change *change)
+{
+	char *uniqueid = ast_strdup(change->uniqueid);
 
 	if (!uniqueid) {
 		ast_log(LOG_ERROR, "Unable to allocate memory for uniqueid\n");
 		return;
 	}
 
-	if (ast_event_get_type(event) != AST_EVENT_UNSUB) {
-		ast_free(uniqueid);
-		return;
-	}
-
-	if (ast_event_get_ie_uint(event, AST_EVENT_IE_EVENTTYPE) != AST_EVENT_MWI) {
-		ast_free(uniqueid);
-		return;
-	}
-
-	u = ast_event_get_ie_uint(event, AST_EVENT_IE_UNIQUEID);
-	*uniqueid = u;
 	if (ast_taskprocessor_push(mwi_subscription_tps, handle_unsubscribe, uniqueid) < 0) {
 		ast_free(uniqueid);
 	}
 }
 
-static void mwi_sub_event_cb(const struct ast_event *event, void *userdata)
+static void mwi_sub_event_cb(struct stasis_subscription_change *change)
 {
 	struct mwi_sub_task *mwist;
+	char *context = ast_strdupa(change->uniqueid);
+	char *mailbox;
 	
-	if (ast_event_get_type(event) != AST_EVENT_SUB)
-		return;
-
-	if (ast_event_get_ie_uint(event, AST_EVENT_IE_EVENTTYPE) != AST_EVENT_MWI)
-		return;
-
 	if ((mwist = ast_calloc(1, (sizeof(*mwist)))) == NULL) {
 		ast_log(LOG_ERROR, "could not allocate a mwi_sub_task\n");
 		return;
 	}
-	mwist->mailbox = ast_strdup(ast_event_get_ie_str(event, AST_EVENT_IE_MAILBOX));
-	mwist->context = ast_strdup(ast_event_get_ie_str(event, AST_EVENT_IE_CONTEXT));
-	mwist->uniqueid = ast_event_get_ie_uint(event, AST_EVENT_IE_UNIQUEID);
+
+	mailbox = strsep(&context, "@");
+
+	mwist->mailbox = ast_strdup(mailbox);
+	mwist->context = ast_strdup(context);
+	mwist->uniqueid = ast_strdup(change->uniqueid);
 	
 	if (ast_taskprocessor_push(mwi_subscription_tps, handle_subscribe, mwist) < 0) {
 		ast_free(mwist);
 	}
 }
 
+static void mwi_event_cb(void *userdata, struct stasis_topic *topic, struct stasis_message *msg)
+{
+	struct stasis_subscription_change *change;
+	/* Only looking for subscription change notices here */
+	if (stasis_message_type(msg) != stasis_subscription_change()) {
+		return;
+	}
+
+	change = stasis_message_data(msg);
+	if (!strcmp(change->description, "Subscribe")) {
+		mwi_sub_event_cb(change);
+	} else if (!strcmp(change->description, "Unsubscribe")) {
+		mwi_unsub_event_cb(change);
+	}
+}
+
+static int dump_cache(void *obj, void *arg, int flags)
+{
+	struct stasis_message *msg = arg;
+	mwi_event_cb(NULL, NULL, msg);
+	return 0;
+}
+
 static void start_poll_thread(void)
 {
 	int errcode;
-	mwi_sub_sub = ast_event_subscribe(AST_EVENT_SUB, mwi_sub_event_cb, "Voicemail MWI subscription", NULL,
-		AST_EVENT_IE_EVENTTYPE, AST_EVENT_IE_PLTYPE_UINT, AST_EVENT_MWI,
-		AST_EVENT_IE_END);
-
-	mwi_unsub_sub = ast_event_subscribe(AST_EVENT_UNSUB, mwi_unsub_event_cb, "Voicemail MWI subscription", NULL,
-		AST_EVENT_IE_EVENTTYPE, AST_EVENT_IE_PLTYPE_UINT, AST_EVENT_MWI,
-		AST_EVENT_IE_END);
-
-	if (mwi_sub_sub)
-		ast_event_report_subs(mwi_sub_sub);
+	mwi_sub_sub = stasis_subscribe(stasis_mwi_topic_all(), mwi_event_cb, NULL);
+
+	if (mwi_sub_sub) {
+		struct ao2_container *cached = stasis_cache_dump(stasis_mwi_topic_cached(), stasis_subscription_change());
+		if (cached) {
+			ao2_callback(cached, OBJ_MULTIPLE | OBJ_NODATA, dump_cache, NULL);
+		}
+		ao2_cleanup(cached);
+	}
 
 	poll_thread_run = 1;
 
@@ -12669,13 +12677,8 @@
 	poll_thread_run = 0;
 
 	if (mwi_sub_sub) {
-		ast_event_unsubscribe(mwi_sub_sub);
+		stasis_unsubscribe(mwi_sub_sub);
 		mwi_sub_sub = NULL;
-	}
-
-	if (mwi_unsub_sub) {
-		ast_event_unsubscribe(mwi_unsub_sub);
-		mwi_unsub_sub = NULL;
 	}
 
 	ast_mutex_lock(&poll_lock);




More information about the asterisk-commits mailing list