kenmirrors/asterisk
1
0
Fork 0
mirror of https://github.com/asterisk/asterisk.git synced 2025-09-29 18:19:30 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move presence state distribution to Stasis-core

Browse Source 
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@digium.com>


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@385862 65c4cc65-6c06-0410-ace0-fbb531ad65f3
This commit is contained in:
Kinsey Moore
2013-04-16 15:48:16 +00:00
parent 191cf99ae1
commit 71a01725b8
4 changed files with 239 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
funcs/func_presencestate.c
View File

@@ -645,21 +645,20 @@ AST_TEST_DEFINE(test_invalid_parse_data)
}
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 @@ static void test_cb(const struct ast_event *event, void *userdata)
*/
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 @@ AST_TEST_DEFINE(test_presence_state_change)
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 @@ AST_TEST_DEFINE(test_presence_state_change)
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");

34
include/asterisk/presencestate.h
View File

@@ -149,6 +149,40 @@ int ast_presence_state_prov_add(const char *label, ast_presence_state_prov_cb_ty
*/
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

264
main/pbx.c
View File

@@ -819,8 +819,6 @@ AST_APP_OPTIONS(waitexten_opts, {
struct ast_context;
struct ast_app;
static struct ast_taskprocessor *extension_state_tps;
AST_THREADSTORAGE(switch_data);
AST_THREADSTORAGE(extensionstate_buf);
@@ -1123,13 +1121,6 @@ static const struct cfextension_states {
{ AST_EXTENSION_INUSE | AST_EXTENSION_ONHOLD, "InUse&Hold" }
};
struct presencechange {
char *provider;
int state;
char *subtype;
char *message;
};
struct pbx_exception {
AST_DECLARE_STRING_FIELDS(
AST_STRING_FIELD(context); /*!< Context associated with this exception */
@@ -1297,7 +1288,7 @@ static char *overrideswitch = NULL;
/*! \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 @@ static int execute_state_callback(ast_state_cb_type cb,
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 @@ static int pbx_builtin_sayphonetic(struct ast_channel *chan, const char *data)
return res;
}
static void presencechange_destroy(void *data)
static void presence_state_cb(void *unused, struct stasis_subscription *sub, struct stasis_topic *topic, struct stasis_message *msg)
{
struct presencechange *pc = data;
ast_free(pc->provider);
ast_free(pc->subtype);
ast_free(pc->message);
}
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];
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))) {
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);
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);
hint_app = ast_str_create(1024);
if (!hint_app) {
return;
}
if ((tmp = ast_event_get_ie_str(event, AST_EVENT_IE_PRESENCE_SUBTYPE))) {
pc->subtype = ast_strdup(tmp);
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;
}
if ((tmp = ast_event_get_ie_str(event, AST_EVENT_IE_PRESENCE_MESSAGE))) {
pc->message = ast_strdup(tmp);
/* 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;
}
/* 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_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 @@ static void unload_pbx(void)
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 @@ static void unload_pbx(void)
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 @@ int load_pbx(void)
/* 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 @@ int load_pbx(void)
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;
}

159
main/presencestate.c
View File

@@ -53,13 +53,9 @@ static const struct {
{ "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 @@ struct presence_state_provider {
/*! \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 @@ enum ast_presence_state ast_presence_state_val(const char *val)
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;
RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
struct ast_presence_state_message *presence_state;
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);
msg = stasis_cache_get(ast_presence_state_topic_cached(), ast_presence_state_message_type(), presence_provider);
if (!event) {
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);
presence_state = stasis_message_data(msg);
res = presence_state->state;
*subtype = !ast_strlen_zero(_subtype) ? ast_strdup(_subtype) : NULL;
*message = !ast_strlen_zero(_message) ? ast_strdup(_message) : NULL;
ast_event_destroy(event);
*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 @@ int ast_presence_state_prov_del(const char *label)
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;
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 (!(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))) {
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 @@ int ast_presence_state_changed_literal(enum ast_presence_state state,
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 @@ int ast_presence_state_changed(enum ast_presence_state state,
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 stasis_message_type *ast_presence_state_message_type(void)
{
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 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;
}