[asterisk-commits] kmoore: trunk r385862 - in /trunk: funcs/ include/asterisk/ main/
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Tue Apr 16 10:48:20 CDT 2013
Author: kmoore
Date: Tue Apr 16 10:48:16 2013
New Revision: 385862
URL: http://svnview.digium.com/svn/asterisk?view=rev&rev=385862
Log:
Move presence state distribution to Stasis-core
Convert presence state events to Stasis-core messages and remove
redundant serializers where possible.
Review: https://reviewboard.asterisk.org/r/2410/
(closes issue ASTERISK-21102)
Patch-by: Kinsey Moore <kmoore at digium.com>
Modified:
trunk/funcs/func_presencestate.c
trunk/include/asterisk/presencestate.h
trunk/main/pbx.c
trunk/main/presencestate.c
Modified: trunk/funcs/func_presencestate.c
URL: http://svnview.digium.com/svn/asterisk/trunk/funcs/func_presencestate.c?view=diff&rev=385862&r1=385861&r2=385862
==============================================================================
--- trunk/funcs/func_presencestate.c (original)
+++ trunk/funcs/func_presencestate.c Tue Apr 16 10:48:16 2013
@@ -645,21 +645,20 @@
}
struct test_cb_data {
- enum ast_presence_state presence;
- const char *provider;
- const char *subtype;
- const char *message;
+ struct ast_presence_state_message *presence_state;
/* That's right. I'm using a semaphore */
sem_t sem;
};
-static void test_cb(const struct ast_event *event, void *userdata)
+static void test_cb(void *userdata, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *msg)
{
struct test_cb_data *cb_data = userdata;
- cb_data->presence = ast_event_get_ie_uint(event, AST_EVENT_IE_PRESENCE_STATE);
- cb_data->provider = ast_strdup(ast_event_get_ie_str(event, AST_EVENT_IE_PRESENCE_PROVIDER));
- cb_data->subtype = ast_strdup(ast_event_get_ie_str(event, AST_EVENT_IE_PRESENCE_SUBTYPE));
- cb_data->message = ast_strdup(ast_event_get_ie_str(event, AST_EVENT_IE_PRESENCE_MESSAGE));
+ if (stasis_message_type(msg) != ast_presence_state_message_type()) {
+ return;
+ }
+ cb_data->presence_state = stasis_message_data(msg);
+ ao2_ref(cb_data->presence_state, +1);
+
sem_post(&cb_data->sem);
}
@@ -670,7 +669,7 @@
*/
AST_TEST_DEFINE(test_presence_state_change)
{
- struct ast_event_sub *test_sub;
+ struct stasis_subscription *test_sub;
struct test_cb_data *cb_data;
switch (cmd) {
@@ -690,8 +689,7 @@
return AST_TEST_FAIL;
}
- if (!(test_sub = ast_event_subscribe(AST_EVENT_PRESENCE_STATE,
- test_cb, "Test presence state callbacks", cb_data, AST_EVENT_IE_END))) {
+ if (!(test_sub = stasis_subscribe(ast_presence_state_topic_all(), test_cb, cb_data))) {
return AST_TEST_FAIL;
}
@@ -701,16 +699,16 @@
presence_write(NULL, "PRESENCESTATE", "CustomPresence:TestPresenceStateChange", "away,down the hall,Quarterly financial meeting");
sem_wait(&cb_data->sem);
- if (cb_data->presence != AST_PRESENCE_AWAY ||
- strcmp(cb_data->provider, "CustomPresence:TestPresenceStateChange") ||
- strcmp(cb_data->subtype, "down the hall") ||
- strcmp(cb_data->message, "Quarterly financial meeting")) {
+ if (cb_data->presence_state->state != AST_PRESENCE_AWAY ||
+ strcmp(cb_data->presence_state->provider, "CustomPresence:TestPresenceStateChange") ||
+ strcmp(cb_data->presence_state->subtype, "down the hall") ||
+ strcmp(cb_data->presence_state->message, "Quarterly financial meeting")) {
return AST_TEST_FAIL;
}
- ast_free((char *)cb_data->provider);
- ast_free((char *)cb_data->subtype);
- ast_free((char *)cb_data->message);
+ test_sub = stasis_unsubscribe(test_sub);
+
+ ao2_cleanup(cb_data->presence_state);
ast_free((char *)cb_data);
ast_db_del("CustomPresence", "TestPresenceStateChange");
Modified: trunk/include/asterisk/presencestate.h
URL: http://svnview.digium.com/svn/asterisk/trunk/include/asterisk/presencestate.h?view=diff&rev=385862&r1=385861&r2=385862
==============================================================================
--- trunk/include/asterisk/presencestate.h (original)
+++ trunk/include/asterisk/presencestate.h Tue Apr 16 10:48:16 2013
@@ -149,6 +149,40 @@
*/
int ast_presence_state_prov_del(const char *label);
+/*!
+ * \brief Get presence state message type
+ * \retval Stasis message type for presence state messages
+ * \since 12
+ */
+struct stasis_message_type *ast_presence_state_message_type(void);
+
+/*!
+ * \brief Get presence state topic
+ * \retval Stasis topic for presence state messages
+ * \since 12
+ */
+struct stasis_topic *ast_presence_state_topic_all(void);
+
+/*!
+ * \brief Get caching presence state topic
+ * \retval Caching Stasis topic for presence state messages
+ * \since 12
+ */
+struct stasis_caching_topic *ast_presence_state_topic_cached(void);
+
+/*!
+ * \brief Stasis message payload representing a presence state update
+ * \since 12
+ */
+struct ast_presence_state_message {
+ AST_DECLARE_STRING_FIELDS(
+ AST_STRING_FIELD(provider); /*!< Provider that produced this presence state message */
+ AST_STRING_FIELD(subtype); /*!< Subtype associated with this presence state message */
+ AST_STRING_FIELD(message); /*!< The message to convey */
+ );
+ enum ast_presence_state state; /*!< The state associated with this presence state message */
+};
+
int ast_presence_state_engine_init(void);
#endif
Modified: trunk/main/pbx.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/pbx.c?view=diff&rev=385862&r1=385861&r2=385862
==============================================================================
--- trunk/main/pbx.c (original)
+++ trunk/main/pbx.c Tue Apr 16 10:48:16 2013
@@ -819,8 +819,6 @@
struct ast_context;
struct ast_app;
-static struct ast_taskprocessor *extension_state_tps;
-
AST_THREADSTORAGE(switch_data);
AST_THREADSTORAGE(extensionstate_buf);
@@ -1121,13 +1119,6 @@
{ AST_EXTENSION_INUSE | AST_EXTENSION_RINGING, "InUse&Ringing" },
{ AST_EXTENSION_ONHOLD, "Hold" },
{ AST_EXTENSION_INUSE | AST_EXTENSION_ONHOLD, "InUse&Hold" }
-};
-
-struct presencechange {
- char *provider;
- int state;
- char *subtype;
- char *message;
};
struct pbx_exception {
@@ -1297,7 +1288,7 @@
/*! \brief Subscription for device state change events */
static struct stasis_subscription *device_state_sub;
/*! \brief Subscription for presence state change events */
-static struct ast_event_sub *presence_state_sub;
+static struct stasis_subscription *presence_state_sub;
AST_MUTEX_DEFINE_STATIC(maxcalllock);
static int countcalls;
@@ -5042,125 +5033,6 @@
return res;
}
-static int handle_presencechange(void *datap)
-{
- struct ast_hint *hint;
- struct ast_str *hint_app = NULL;
- struct presencechange *pc = datap;
- struct ao2_iterator i;
- struct ao2_iterator cb_iter;
- char context_name[AST_MAX_CONTEXT];
- char exten_name[AST_MAX_EXTENSION];
- int res = -1;
-
- hint_app = ast_str_create(1024);
- if (!hint_app) {
- goto presencechange_cleanup;
- }
-
- ast_mutex_lock(&context_merge_lock);/* Hold off ast_merge_contexts_and_delete */
- i = ao2_iterator_init(hints, 0);
- for (; (hint = ao2_iterator_next(&i)); ao2_ref(hint, -1)) {
- struct ast_state_cb *state_cb;
- const char *app;
- char *parse;
-
- ao2_lock(hint);
-
- if (!hint->exten) {
- /* The extension has already been destroyed */
- ao2_unlock(hint);
- continue;
- }
-
- /* Does this hint monitor the device that changed state? */
- app = ast_get_extension_app(hint->exten);
- if (ast_strlen_zero(app)) {
- /* The hint does not monitor presence at all. */
- ao2_unlock(hint);
- continue;
- }
-
- ast_str_set(&hint_app, 0, "%s", app);
- parse = parse_hint_presence(hint_app);
- if (ast_strlen_zero(parse)) {
- ao2_unlock(hint);
- continue;
- }
- if (strcasecmp(parse, pc->provider)) {
- /* The hint does not monitor the presence provider. */
- ao2_unlock(hint);
- continue;
- }
-
- /*
- * Save off strings in case the hint extension gets destroyed
- * while we are notifying the watchers.
- */
- ast_copy_string(context_name,
- ast_get_context_name(ast_get_extension_context(hint->exten)),
- sizeof(context_name));
- ast_copy_string(exten_name, ast_get_extension_name(hint->exten),
- sizeof(exten_name));
- ast_str_set(&hint_app, 0, "%s", ast_get_extension_app(hint->exten));
-
- /* Check to see if update is necessary */
- if ((hint->last_presence_state == pc->state) &&
- ((hint->last_presence_subtype && pc->subtype && !strcmp(hint->last_presence_subtype, pc->subtype)) || (!hint->last_presence_subtype && !pc->subtype)) &&
- ((hint->last_presence_message && pc->message && !strcmp(hint->last_presence_message, pc->message)) || (!hint->last_presence_message && !pc->message))) {
-
- /* this update is the same as the last, do nothing */
- ao2_unlock(hint);
- continue;
- }
-
- /* update new values */
- ast_free(hint->last_presence_subtype);
- ast_free(hint->last_presence_message);
- hint->last_presence_state = pc->state;
- hint->last_presence_subtype = pc->subtype ? ast_strdup(pc->subtype) : NULL;
- hint->last_presence_message = pc->message ? ast_strdup(pc->message) : NULL;
-
- ao2_unlock(hint);
-
- /* For general callbacks */
- cb_iter = ao2_iterator_init(statecbs, 0);
- for (; (state_cb = ao2_iterator_next(&cb_iter)); ao2_ref(state_cb, -1)) {
- execute_state_callback(state_cb->change_cb,
- context_name,
- exten_name,
- state_cb->data,
- AST_HINT_UPDATE_PRESENCE,
- hint,
- NULL);
- }
- ao2_iterator_destroy(&cb_iter);
-
- /* For extension callbacks */
- cb_iter = ao2_iterator_init(hint->callbacks, 0);
- for (; (state_cb = ao2_iterator_next(&cb_iter)); ao2_ref(state_cb, -1)) {
- execute_state_callback(state_cb->change_cb,
- context_name,
- exten_name,
- state_cb->data,
- AST_HINT_UPDATE_PRESENCE,
- hint,
- NULL);
- }
- ao2_iterator_destroy(&cb_iter);
- }
- ao2_iterator_destroy(&i);
- ast_mutex_unlock(&context_merge_lock);
-
- res = 0;
-
-presencechange_cleanup:
- ast_free(hint_app);
- ao2_ref(pc, -1);
-
- return res;
-}
-
/*!
* /internal
* /brief Identify a channel for every device which is supposedly responsible for the device state.
@@ -11663,49 +11535,112 @@
return res;
}
-static void presencechange_destroy(void *data)
-{
- struct presencechange *pc = data;
- ast_free(pc->provider);
- ast_free(pc->subtype);
- ast_free(pc->message);
-}
-
-static void presence_state_cb(const struct ast_event *event, void *unused)
-{
- struct presencechange *pc;
- const char *tmp;
-
- if (!(pc = ao2_alloc(sizeof(*pc), presencechange_destroy))) {
+static void presence_state_cb(void *unused, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *msg)
+{
+ struct ast_presence_state_message *presence_state = stasis_message_data(msg);
+ struct ast_hint *hint;
+ struct ast_str *hint_app = NULL;
+ struct ao2_iterator hint_iter;
+ struct ao2_iterator cb_iter;
+ char context_name[AST_MAX_CONTEXT];
+ char exten_name[AST_MAX_EXTENSION];
+
+ if (stasis_message_type(msg) != ast_presence_state_message_type()) {
return;
}
- tmp = ast_event_get_ie_str(event, AST_EVENT_IE_PRESENCE_PROVIDER);
- if (ast_strlen_zero(tmp)) {
- ast_log(LOG_ERROR, "Received invalid event that had no presence provider IE\n");
- ao2_ref(pc, -1);
+ hint_app = ast_str_create(1024);
+ if (!hint_app) {
return;
}
- pc->provider = ast_strdup(tmp);
-
- pc->state = ast_event_get_ie_uint(event, AST_EVENT_IE_PRESENCE_STATE);
- if (pc->state < 0) {
- ao2_ref(pc, -1);
- return;
- }
-
- if ((tmp = ast_event_get_ie_str(event, AST_EVENT_IE_PRESENCE_SUBTYPE))) {
- pc->subtype = ast_strdup(tmp);
- }
-
- if ((tmp = ast_event_get_ie_str(event, AST_EVENT_IE_PRESENCE_MESSAGE))) {
- pc->message = ast_strdup(tmp);
- }
-
- /* The task processor thread is taking our reference to the presencechange object. */
- if (ast_taskprocessor_push(extension_state_tps, handle_presencechange, pc) < 0) {
- ao2_ref(pc, -1);
- }
+
+ ast_mutex_lock(&context_merge_lock);/* Hold off ast_merge_contexts_and_delete */
+ hint_iter = ao2_iterator_init(hints, 0);
+ for (; (hint = ao2_iterator_next(&hint_iter)); ao2_cleanup(hint)) {
+ struct ast_state_cb *state_cb;
+ const char *app;
+ char *parse;
+ SCOPED_AO2LOCK(lock, hint);
+
+ if (!hint->exten) {
+ /* The extension has already been destroyed */
+ continue;
+ }
+
+ /* Does this hint monitor the device that changed state? */
+ app = ast_get_extension_app(hint->exten);
+ if (ast_strlen_zero(app)) {
+ /* The hint does not monitor presence at all. */
+ continue;
+ }
+
+ ast_str_set(&hint_app, 0, "%s", app);
+ parse = parse_hint_presence(hint_app);
+ if (ast_strlen_zero(parse)) {
+ continue;
+ }
+ if (strcasecmp(parse, presence_state->provider)) {
+ /* The hint does not monitor the presence provider. */
+ continue;
+ }
+
+ /*
+ * Save off strings in case the hint extension gets destroyed
+ * while we are notifying the watchers.
+ */
+ ast_copy_string(context_name,
+ ast_get_context_name(ast_get_extension_context(hint->exten)),
+ sizeof(context_name));
+ ast_copy_string(exten_name, ast_get_extension_name(hint->exten),
+ sizeof(exten_name));
+ ast_str_set(&hint_app, 0, "%s", ast_get_extension_app(hint->exten));
+
+ /* Check to see if update is necessary */
+ if ((hint->last_presence_state == presence_state->state) &&
+ ((hint->last_presence_subtype && presence_state->subtype && !strcmp(hint->last_presence_subtype, presence_state->subtype)) || (!hint->last_presence_subtype && !presence_state->subtype)) &&
+ ((hint->last_presence_message && presence_state->message && !strcmp(hint->last_presence_message, presence_state->message)) || (!hint->last_presence_message && !presence_state->message))) {
+
+ /* this update is the same as the last, do nothing */
+ continue;
+ }
+
+ /* update new values */
+ ast_free(hint->last_presence_subtype);
+ ast_free(hint->last_presence_message);
+ hint->last_presence_state = presence_state->state;
+ hint->last_presence_subtype = presence_state->subtype ? ast_strdup(presence_state->subtype) : NULL;
+ hint->last_presence_message = presence_state->message ? ast_strdup(presence_state->message) : NULL;
+
+ /* For general callbacks */
+ cb_iter = ao2_iterator_init(statecbs, 0);
+ for (; (state_cb = ao2_iterator_next(&cb_iter)); ao2_ref(state_cb, -1)) {
+ execute_state_callback(state_cb->change_cb,
+ context_name,
+ exten_name,
+ state_cb->data,
+ AST_HINT_UPDATE_PRESENCE,
+ hint,
+ NULL);
+ }
+ ao2_iterator_destroy(&cb_iter);
+
+ /* For extension callbacks */
+ cb_iter = ao2_iterator_init(hint->callbacks, 0);
+ for (; (state_cb = ao2_iterator_next(&cb_iter)); ao2_cleanup(state_cb)) {
+ execute_state_callback(state_cb->change_cb,
+ context_name,
+ exten_name,
+ state_cb->data,
+ AST_HINT_UPDATE_PRESENCE,
+ hint,
+ NULL);
+ }
+ ao2_iterator_destroy(&cb_iter);
+ }
+ ao2_iterator_destroy(&hint_iter);
+ ast_mutex_unlock(&context_merge_lock);
+
+ ast_free(hint_app);
}
/*!
@@ -11765,7 +11700,7 @@
int x;
if (presence_state_sub) {
- presence_state_sub = ast_event_unsubscribe(presence_state_sub);
+ presence_state_sub = stasis_unsubscribe(presence_state_sub);
}
if (device_state_sub) {
device_state_sub = stasis_unsubscribe(device_state_sub);
@@ -11780,9 +11715,6 @@
ast_custom_function_unregister(&exception_function);
ast_custom_function_unregister(&testtime_function);
ast_data_unregister(NULL);
- if (extension_state_tps) {
- extension_state_tps = ast_taskprocessor_unreference(extension_state_tps);
- }
}
int load_pbx(void)
@@ -11793,9 +11725,6 @@
/* Initialize the PBX */
ast_verb(1, "Asterisk PBX Core Initializing\n");
- if (!(extension_state_tps = ast_taskprocessor_get("pbx-core", 0))) {
- ast_log(LOG_WARNING, "failed to create pbx-core taskprocessor\n");
- }
ast_verb(1, "Registering builtin applications:\n");
ast_cli_register_multiple(pbx_cli, ARRAY_LEN(pbx_cli));
@@ -11819,8 +11748,7 @@
return -1;
}
- if (!(presence_state_sub = ast_event_subscribe(AST_EVENT_PRESENCE_STATE, presence_state_cb, "pbx Presence State Change", NULL,
- AST_EVENT_IE_END))) {
+ if (!(presence_state_sub = stasis_subscribe(ast_presence_state_topic_all(), presence_state_cb, NULL))) {
return -1;
}
Modified: trunk/main/presencestate.c
URL: http://svnview.digium.com/svn/asterisk/trunk/main/presencestate.c?view=diff&rev=385862&r1=385861&r2=385862
==============================================================================
--- trunk/main/presencestate.c (original)
+++ trunk/main/presencestate.c Tue Apr 16 10:48:16 2013
@@ -53,13 +53,9 @@
{ "dnd", AST_PRESENCE_DND},
};
-/*! \brief Flag for the queue */
-static ast_cond_t change_pending;
-
-struct state_change {
- AST_LIST_ENTRY(state_change) list;
- char provider[1];
-};
+struct stasis_message_type *presence_state_type;
+struct stasis_topic *presence_state_topic_all;
+struct stasis_caching_topic *presence_state_topic_cached;
/*! \brief A presence state provider */
struct presence_state_provider {
@@ -71,13 +67,6 @@
/*! \brief A list of providers */
static AST_RWLIST_HEAD_STATIC(presence_state_providers, presence_state_provider);
-/*! \brief The state change queue. State changes are queued
- for processing by a separate thread */
-static AST_LIST_HEAD_STATIC(state_changes, state_change);
-
-/*! \brief The presence state change notification thread */
-static pthread_t change_thread = AST_PTHREADT_NULL;
-
const char *ast_presence_state2str(enum ast_presence_state state)
{
int i;
@@ -103,25 +92,20 @@
static enum ast_presence_state presence_state_cached(const char *presence_provider, char **subtype, char **message)
{
enum ast_presence_state res = AST_PRESENCE_INVALID;
- struct ast_event *event;
- const char *_subtype;
- const char *_message;
-
- event = ast_event_get_cached(AST_EVENT_PRESENCE_STATE,
- AST_EVENT_IE_PRESENCE_PROVIDER, AST_EVENT_IE_PLTYPE_STR, presence_provider,
- AST_EVENT_IE_END);
-
- if (!event) {
+ RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+ struct ast_presence_state_message *presence_state;
+
+ msg = stasis_cache_get(ast_presence_state_topic_cached(), ast_presence_state_message_type(), presence_provider);
+
+ if (!msg) {
return res;
}
- res = ast_event_get_ie_uint(event, AST_EVENT_IE_PRESENCE_STATE);
- _subtype = ast_event_get_ie_str(event, AST_EVENT_IE_PRESENCE_SUBTYPE);
- _message = ast_event_get_ie_str(event, AST_EVENT_IE_PRESENCE_MESSAGE);
-
- *subtype = !ast_strlen_zero(_subtype) ? ast_strdup(_subtype) : NULL;
- *message = !ast_strlen_zero(_message) ? ast_strdup(_message) : NULL;
- ast_event_destroy(event);
+ presence_state = stasis_message_data(msg);
+ res = presence_state->state;
+
+ *subtype = !ast_strlen_zero(presence_state->subtype) ? ast_strdup(presence_state->subtype) : NULL;
+ *message = !ast_strlen_zero(presence_state->message) ? ast_strdup(presence_state->message) : NULL;
return res;
}
@@ -213,23 +197,50 @@
return res;
}
+static void presence_state_dtor(void *obj)
+{
+ struct ast_presence_state_message *presence_state = obj;
+ ast_string_field_free_memory(presence_state);
+}
+
+static struct ast_presence_state_message *presence_state_alloc(const char *provider,
+ enum ast_presence_state state,
+ const char *subtype,
+ const char *message)
+{
+ RAII_VAR(struct ast_presence_state_message *, presence_state, ao2_alloc(sizeof(*presence_state), presence_state_dtor), ao2_cleanup);
+
+ if (!presence_state || ast_string_field_init(presence_state, 256)) {
+ return NULL;
+ }
+
+ presence_state->state = state;
+ ast_string_field_set(presence_state, provider, provider);
+ ast_string_field_set(presence_state, subtype, S_OR(subtype, ""));
+ ast_string_field_set(presence_state, message, S_OR(message, ""));
+
+ ao2_ref(presence_state, +1);
+ return presence_state;
+}
+
static void presence_state_event(const char *provider,
enum ast_presence_state state,
const char *subtype,
const char *message)
{
- struct ast_event *event;
-
- if (!(event = ast_event_new(AST_EVENT_PRESENCE_STATE,
- AST_EVENT_IE_PRESENCE_PROVIDER, AST_EVENT_IE_PLTYPE_STR, provider,
- AST_EVENT_IE_PRESENCE_STATE, AST_EVENT_IE_PLTYPE_UINT, state,
- AST_EVENT_IE_PRESENCE_SUBTYPE, AST_EVENT_IE_PLTYPE_STR, S_OR(subtype, ""),
- AST_EVENT_IE_PRESENCE_MESSAGE, AST_EVENT_IE_PLTYPE_STR, S_OR(message, ""),
- AST_EVENT_IE_END))) {
+ RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
+ RAII_VAR(struct ast_presence_state_message *, presence_state, presence_state_alloc(provider, state, subtype, message), ao2_cleanup);
+
+ if (!presence_state) {
return;
}
- ast_event_queue_and_cache(event);
+ msg = stasis_message_create(ast_presence_state_message_type(), presence_state);
+ if (!msg) {
+ return;
+ }
+
+ stasis_publish(ast_presence_state_topic_all(), msg);
}
static void do_presence_state_change(const char *provider)
@@ -254,19 +265,10 @@
const char *message,
const char *presence_provider)
{
- struct state_change *change;
-
- if (state != AST_PRESENCE_NOT_SET) {
- presence_state_event(presence_provider, state, subtype, message);
- } else if ((change_thread == AST_PTHREADT_NULL) ||
- !(change = ast_calloc(1, sizeof(*change) + strlen(presence_provider)))) {
+ if (state == AST_PRESENCE_NOT_SET) {
do_presence_state_change(presence_provider);
} else {
- strcpy(change->provider, presence_provider);
- AST_LIST_LOCK(&state_changes);
- AST_LIST_INSERT_TAIL(&state_changes, change, list);
- ast_cond_signal(&change_pending);
- AST_LIST_UNLOCK(&state_changes);
+ presence_state_event(presence_provider, state, subtype, message);
}
return 0;
@@ -287,39 +289,60 @@
return ast_presence_state_changed_literal(state, subtype, message, buf);
}
-/*! \brief Go through the presence state change queue and update changes in the presence state thread */
-static void *do_presence_changes(void *data)
-{
- struct state_change *next, *current;
-
- for (;;) {
- /* This basically pops off any state change entries, resets the list back to NULL, unlocks, and processes each state change */
- AST_LIST_LOCK(&state_changes);
- if (AST_LIST_EMPTY(&state_changes))
- ast_cond_wait(&change_pending, &state_changes.lock);
- next = AST_LIST_FIRST(&state_changes);
- AST_LIST_HEAD_INIT_NOLOCK(&state_changes);
- AST_LIST_UNLOCK(&state_changes);
-
- /* Process each state change */
- while ((current = next)) {
- next = AST_LIST_NEXT(current, list);
- do_presence_state_change(current->provider);
- ast_free(current);
- }
- }
-
- return NULL;
+struct stasis_message_type *ast_presence_state_message_type(void)
+{
+ return presence_state_type;
+}
+
+struct stasis_topic *ast_presence_state_topic_all(void)
+{
+ return presence_state_topic_all;
+}
+
+struct stasis_caching_topic *ast_presence_state_topic_cached(void)
+{
+ return presence_state_topic_cached;
+}
+
+static const char *presence_state_get_id(struct stasis_message *msg)
+{
+ struct ast_presence_state_message *presence_state = stasis_message_data(msg);
+
+ if (stasis_message_type(msg) != ast_presence_state_message_type()) {
+ return NULL;
+ }
+
+ return presence_state->provider;
+}
+
+static void presence_state_engine_cleanup(void)
+{
+ ao2_cleanup(presence_state_topic_all);
+ presence_state_topic_all = NULL;
+ ao2_cleanup(presence_state_topic_cached);
+ presence_state_topic_cached = NULL;
+ ao2_cleanup(presence_state_type);
+ presence_state_type = NULL;
}
int ast_presence_state_engine_init(void)
{
- ast_cond_init(&change_pending, NULL);
- if (ast_pthread_create_background(&change_thread, NULL, do_presence_changes, NULL) < 0) {
- ast_log(LOG_ERROR, "Unable to start presence state change thread.\n");
+ presence_state_type = stasis_message_type_create("ast_presence_state_message");
+ if (!presence_state_type) {
return -1;
}
+ presence_state_topic_all = stasis_topic_create("ast_presence_state_topic_all");
+ if (!presence_state_topic_all) {
+ return -1;
+ }
+
+ presence_state_topic_cached = stasis_caching_topic_create(presence_state_topic_all, presence_state_get_id);
+ if (!presence_state_topic_cached) {
+ return -1;
+ }
+ ast_register_atexit(presence_state_engine_cleanup);
+
return 0;
}
More information about the asterisk-commits
mailing list