messaging.c File Reference

Stasis out-of-call text message support. More...

#include "asterisk.h"
#include "asterisk/message.h"
#include "asterisk/endpoints.h"
#include "asterisk/astobj2.h"
#include "asterisk/vector.h"
#include "asterisk/lock.h"
#include "asterisk/utils.h"
#include "asterisk/test.h"
#include "messaging.h"

Include dependency graph for messaging.c:

Go to the source code of this file.

Data Structures
struct	application_tuple
	Storage object for an application. More...
struct	message_subscription
	A subscription to some endpoint or technology. More...

Macros
#define	ENDPOINTS_NUM_BUCKETS   127
	Number of buckets for the endpoint_subscriptions container.
#define	TECH_WILDCARD   "__AST_ALL_TECH"
	Subscription to all technologies.

Functions
static struct application_tuple *	application_tuple_alloc (const char *app_name, message_received_cb callback, void *pvt)
static int	application_tuple_cmp (struct application_tuple *item, const char *key)
static void	application_tuple_dtor (void *obj)
static void	dispatch_message (struct message_subscription *sub, const char *endpoint_name, struct ast_json *json_msg)
static struct message_subscription *	get_or_create_subscription (struct ast_endpoint *endpoint)
static struct message_subscription *	get_subscription (struct ast_endpoint *endpoint)
static int	handle_msg_cb (struct ast_msg *msg)
static int	has_destination_cb (const struct ast_msg *msg)
static int	is_app_subscribed (struct message_subscription *sub, const char *app_name)
static struct message_subscription *	message_subscription_alloc (const char *token)
static int	message_subscription_compare_cb (void *obj, void *arg, int flags)
static void	message_subscription_dtor (void *obj)
static int	message_subscription_hash_cb (const void *obj, const int flags)
int	messaging_app_subscribe_endpoint (const char *app_name, struct ast_endpoint *endpoint, message_received_cb callback, void *pvt)
	Subscribe an application to an endpoint for messages.
void	messaging_app_unsubscribe_endpoint (const char *app_name, const char *endpoint_id)
	Subscribe for messages from a particular endpoint.
int	messaging_cleanup (void)
	Tidy up the messaging layer.
int	messaging_init (void)
	Initialize the messaging layer.
static int	messaging_subscription_cmp (struct message_subscription *sub, const char *key)
static void	msg_to_endpoint (const struct ast_msg *msg, char *buf, size_t len)
static struct ast_json *	msg_to_json (struct ast_msg *msg)

Variables
struct ast_msg_handler	ari_msg_handler
static struct ao2_container *	endpoint_subscriptions
	The subscriptions to endpoints.
struct {
size_t   current
struct message_subscription **   elems
size_t   max
}	tech_subscriptions
	The subscriptions to technologies.
static ast_rwlock_t	tech_subscriptions_lock
	RWLock for tech_subscriptions.

Detailed Description

Stasis out-of-call text message support.

Author

Matt Jordan mjord.nosp@m.an@d.nosp@m.igium.nosp@m..com

Definition in file messaging.c.

Macro Definition Documentation

ENDPOINTS_NUM_BUCKETS

#define ENDPOINTS_NUM_BUCKETS   127

Number of buckets for the endpoint_subscriptions container.

Definition at line 46 of file messaging.c.

TECH_WILDCARD

#define TECH_WILDCARD   "__AST_ALL_TECH"

Subscription to all technologies.

Definition at line 41 of file messaging.c.

Function Documentation

application_tuple_alloc()

static struct application_tuple * application_tuple_alloc ( const char *  app_name, message_received_cb  callback, void *  pvt  )

static

Definition at line 89 of file messaging.c.

{
    struct application_tuple *tuple;
    size_t size = sizeof(*tuple) + strlen(app_name) + 1;
 
    ast_assert(callback != NULL);
 
    tuple = ao2_alloc_options(size, application_tuple_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK);
    if (!tuple) {
        return NULL;
    }
 
    strcpy(tuple->app_name, app_name); /* Safe */
    tuple->pvt = ao2_bump(pvt);
    tuple->callback = callback;
 
    return tuple;
}

References AO2_ALLOC_OPT_LOCK_NOLOCK, ao2_alloc_options, ao2_bump, app_name(), application_tuple::app_name, application_tuple_dtor(), ast_assert, callback(), application_tuple::callback, NULL, and application_tuple::pvt.

Referenced by messaging_app_subscribe_endpoint().

application_tuple_cmp()

static int application_tuple_cmp ( struct application_tuple *  item, const char *  key  )

static

Definition at line 373 of file messaging.c.

{
    return !strcmp(item->app_name, key) ? 1 : 0;
}

References item.

Referenced by messaging_app_unsubscribe_endpoint().

application_tuple_dtor()

static void application_tuple_dtor ( void *  obj )

static

Definition at line 81 of file messaging.c.

{
    struct application_tuple *tuple = obj;
 
    ao2_cleanup(tuple->pvt);
}

References ao2_cleanup, and application_tuple::pvt.

Referenced by application_tuple_alloc().

dispatch_message()

static void dispatch_message ( struct message_subscription *  sub, const char *  endpoint_name, struct ast_json *  json_msg  )

static

Definition at line 292 of file messaging.c.

{
    int i;
 
    ast_debug(3, "Dispatching message to subscription %s for endpoint %s\n",
        sub->token,
        endpoint_name);
    for (i = 0; i < AST_VECTOR_SIZE(&sub->applications); i++) {
        struct application_tuple *tuple = AST_VECTOR_GET(&sub->applications, i);
 
        tuple->callback(endpoint_name, json_msg, tuple->pvt);
    }
}

References ast_debug, AST_VECTOR_GET, AST_VECTOR_SIZE, application_tuple::callback, application_tuple::pvt, and sub.

Referenced by handle_msg_cb().

get_or_create_subscription()

static struct message_subscription * get_or_create_subscription ( struct ast_endpoint *  endpoint )

static

Definition at line 459 of file messaging.c.

{
    struct message_subscription *sub = get_subscription(endpoint);
 
    if (sub) {
        return sub;
    }
 
    sub = message_subscription_alloc(endpoint ? ast_endpoint_get_id(endpoint) : TECH_WILDCARD);
    if (!sub) {
        return NULL;
    }
 
    /* Either endpoint_subscriptions or tech_subscriptions will hold a reference to
     * the subscription. This reference is released to allow the subscription to
     * eventually destruct when there are no longer any applications receiving
     * events from the subscription.
     */
    if (endpoint && !ast_strlen_zero(ast_endpoint_get_resource(endpoint))) {
        ao2_link(endpoint_subscriptions, sub);
    } else {
        ast_rwlock_wrlock(&tech_subscriptions_lock);
        ao2_ref(sub, +1);
        if (AST_VECTOR_APPEND(&tech_subscriptions, sub)) {
            /* Release the refs that were for the vector and the allocation. */
            ao2_ref(sub, -2);
            sub = NULL;
        }
        ast_rwlock_unlock(&tech_subscriptions_lock);
    }
 
    return sub;
}

References ao2_link, ao2_ref, ast_endpoint_get_id(), ast_endpoint_get_resource(), ast_rwlock_unlock, ast_rwlock_wrlock, ast_strlen_zero(), AST_VECTOR_APPEND, endpoint_subscriptions, get_subscription(), message_subscription_alloc(), NULL, sub, tech_subscriptions, tech_subscriptions_lock, and TECH_WILDCARD.

Referenced by messaging_app_subscribe_endpoint().

get_subscription()

static struct message_subscription * get_subscription ( struct ast_endpoint *  endpoint )

static

Definition at line 394 of file messaging.c.

{
    struct message_subscription *sub = NULL;
 
    if (endpoint && !ast_strlen_zero(ast_endpoint_get_resource(endpoint))) {
        sub = ao2_find(endpoint_subscriptions, ast_endpoint_get_id(endpoint), OBJ_SEARCH_KEY);
    } else {
        int i;
 
        ast_rwlock_rdlock(&tech_subscriptions_lock);
        for (i = 0; i < AST_VECTOR_SIZE(&tech_subscriptions); i++) {
            sub = AST_VECTOR_GET(&tech_subscriptions, i);
 
            if (sub && !strcmp(sub->token, endpoint ? ast_endpoint_get_tech(endpoint) : TECH_WILDCARD)) {
                ao2_bump(sub);
                break;
            }
 
            /* Need to reset the pointer at this line to prevent from using the wrong subscription due to
             * the token check failing.
             */
            sub = NULL;
        }
        ast_rwlock_unlock(&tech_subscriptions_lock);
    }
 
    return sub;
}

References ao2_bump, ao2_find, ast_endpoint_get_id(), ast_endpoint_get_resource(), ast_endpoint_get_tech(), ast_rwlock_rdlock, ast_rwlock_unlock, ast_strlen_zero(), AST_VECTOR_GET, AST_VECTOR_SIZE, endpoint_subscriptions, NULL, OBJ_SEARCH_KEY, sub, tech_subscriptions, tech_subscriptions_lock, and TECH_WILDCARD.

Referenced by get_or_create_subscription(), and messaging_app_unsubscribe_endpoint().

handle_msg_cb()

static int handle_msg_cb ( struct ast_msg *  msg )

static

Definition at line 306 of file messaging.c.

{
    /* We have at most 3 subscriptions: TECH_WILDCARD, tech itself, and endpoint. */
    struct message_subscription *matching_subscriptions[3];
    struct message_subscription *sub;
    int i, j;
    int result;
    char buf[256];
    const char *endpoint_name;
    struct ast_json *json_msg;
 
    msg_to_endpoint(msg, buf, sizeof(buf));
    endpoint_name = buf;
    json_msg = msg_to_json(msg);
    if (!json_msg) {
        return -1;
    }
    result = -1;
 
    /* Find subscriptions to TECH_WILDCARD and to the endpoint's technology. */
    ast_rwlock_rdlock(&tech_subscriptions_lock);
    for (i = 0, j = 0; i < AST_VECTOR_SIZE(&tech_subscriptions) && j < 2; i++) {
        sub = AST_VECTOR_GET(&tech_subscriptions, i);
 
        if (!sub) {
            continue;
        }
 
        if (!strcmp(sub->token, TECH_WILDCARD)
            || !strncasecmp(sub->token, buf, strlen(sub->token))) {
            ao2_ref(sub, +1);
            matching_subscriptions[j++] = sub;
        }
    }
    ast_rwlock_unlock(&tech_subscriptions_lock);
 
    /* Find the subscription to this particular endpoint. */
    sub = ao2_find(endpoint_subscriptions, buf, OBJ_SEARCH_KEY);
    if (sub) {
        matching_subscriptions[j++] = sub;
    }
 
    /* Dispatch the message to all matching subscriptions. */
    for (i = 0; i < j; i++) {
        sub = matching_subscriptions[i];
 
        dispatch_message(sub, endpoint_name, json_msg);
 
        ao2_ref(sub, -1);
        result = 0;
    }
 
    ast_json_unref(json_msg);
    return result;
}

References ao2_find, ao2_ref, ast_json_unref(), ast_rwlock_rdlock, ast_rwlock_unlock, AST_VECTOR_GET, AST_VECTOR_SIZE, buf, dispatch_message(), endpoint_subscriptions, msg_to_endpoint(), msg_to_json(), OBJ_SEARCH_KEY, result, sub, tech_subscriptions, tech_subscriptions_lock, and TECH_WILDCARD.

has_destination_cb()

static int has_destination_cb ( const struct ast_msg *  msg )

static

Definition at line 213 of file messaging.c.

{
    struct message_subscription *sub;
    int i;
    char buf[256];
 
    msg_to_endpoint(msg, buf, sizeof(buf));
 
    ast_rwlock_rdlock(&tech_subscriptions_lock);
    for (i = 0; i < AST_VECTOR_SIZE(&tech_subscriptions); i++) {
        sub = AST_VECTOR_GET(&tech_subscriptions, i);
 
        if (!sub) {
            continue;
        }
 
        if (!strcmp(sub->token, TECH_WILDCARD)
            || !strncasecmp(sub->token, buf, strlen(sub->token))
            || !strncasecmp(sub->token, buf, strlen(sub->token))) {
            ast_rwlock_unlock(&tech_subscriptions_lock);
            goto match;
        }
 
    }
    ast_rwlock_unlock(&tech_subscriptions_lock);
 
    sub = ao2_find(endpoint_subscriptions, buf, OBJ_SEARCH_KEY);
    if (sub) {
        ao2_ref(sub, -1);
        goto match;
    }
 
    ast_debug(1, "No subscription found for %s\n", buf);
    return 0;
 
match:
    return 1;
}

References ao2_find, ao2_ref, ast_debug, ast_rwlock_rdlock, ast_rwlock_unlock, AST_VECTOR_GET, AST_VECTOR_SIZE, buf, endpoint_subscriptions, match(), msg_to_endpoint(), OBJ_SEARCH_KEY, sub, tech_subscriptions, tech_subscriptions_lock, and TECH_WILDCARD.

is_app_subscribed()

static int is_app_subscribed ( struct message_subscription *  sub, const char *  app_name  )

static

Definition at line 378 of file messaging.c.

{
    int i;
 
    for (i = 0; i < AST_VECTOR_SIZE(&sub->applications); i++) {
        struct application_tuple *tuple;
 
        tuple = AST_VECTOR_GET(&sub->applications, i);
        if (tuple && !strcmp(tuple->app_name, app_name)) {
            return 1;
        }
    }
 
    return 0;
}

References app_name(), application_tuple::app_name, AST_VECTOR_GET, AST_VECTOR_SIZE, and sub.

Referenced by messaging_app_subscribe_endpoint(), and messaging_app_unsubscribe_endpoint().

message_subscription_alloc()

static struct message_subscription * message_subscription_alloc ( const char *  token )

static

Definition at line 123 of file messaging.c.

{
    struct message_subscription *sub;
    size_t size = sizeof(*sub) + strlen(token) + 1;
 
    sub = ao2_alloc_options(size, message_subscription_dtor, AO2_ALLOC_OPT_LOCK_RWLOCK);
    if (!sub) {
        return NULL;
    }
    strcpy(sub->token, token); /* Safe */
 
    return sub;
}

References AO2_ALLOC_OPT_LOCK_RWLOCK, ao2_alloc_options, message_subscription_dtor(), NULL, sub, and message_subscription::token.

Referenced by get_or_create_subscription().

message_subscription_compare_cb()

static int message_subscription_compare_cb ( void *  obj, void *  arg, int  flags  )

static

AO2 comparison function for message_subscription

Definition at line 160 of file messaging.c.

{
    const struct message_subscription *object_left = obj;
    const struct message_subscription *object_right = arg;
    const char *right_key = arg;
    int cmp;
 
    switch (flags & OBJ_SEARCH_MASK) {
    case OBJ_SEARCH_OBJECT:
        right_key = object_right->token;
        /* Fall through */
    case OBJ_SEARCH_KEY:
        cmp = strcmp(object_left->token, right_key);
        break;
    case OBJ_SEARCH_PARTIAL_KEY:
        /*
         * We could also use a partial key struct containing a length
         * so strlen() does not get called for every comparison instead.
         */
        cmp = strncmp(object_left->token, right_key, strlen(right_key));
        break;
    default:
        /*
         * What arg points to is specific to this traversal callback
         * and has no special meaning to astobj2.
         */
        cmp = 0;
        break;
    }
    if (cmp) {
        return 0;
    }
    /*
     * At this point the traversal callback is identical to a sorted
     * container.
     */
    return CMP_MATCH;
}

References CMP_MATCH, OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, OBJ_SEARCH_OBJECT, OBJ_SEARCH_PARTIAL_KEY, and message_subscription::token.

Referenced by messaging_init().

message_subscription_dtor()

static void message_subscription_dtor ( void *  obj )

static

Definition at line 109 of file messaging.c.

{
    struct message_subscription *sub = obj;
    int i;
 
    for (i = 0; i < AST_VECTOR_SIZE(&sub->applications); i++) {
        struct application_tuple *tuple = AST_VECTOR_GET(&sub->applications, i);
 
        ao2_cleanup(tuple);
    }
    AST_VECTOR_FREE(&sub->applications);
}

References ao2_cleanup, AST_VECTOR_FREE, AST_VECTOR_GET, AST_VECTOR_SIZE, and sub.

Referenced by message_subscription_alloc().

message_subscription_hash_cb()

static int message_subscription_hash_cb ( const void *  obj, const int  flags  )

static

AO2 hash function for message_subscription

Definition at line 138 of file messaging.c.

{
    const struct message_subscription *sub;
    const char *key;
 
    switch (flags & OBJ_SEARCH_MASK) {
    case OBJ_SEARCH_KEY:
        key = obj;
        break;
    case OBJ_SEARCH_OBJECT:
        sub = obj;
        key = sub->token;
        break;
    default:
        /* Hash can only work on something with a full key. */
        ast_assert(0);
        return 0;
    }
    return ast_str_hash(key);
}

References ast_assert, ast_str_hash(), OBJ_SEARCH_KEY, OBJ_SEARCH_MASK, OBJ_SEARCH_OBJECT, and sub.

Referenced by messaging_init().

messaging_app_subscribe_endpoint()

int messaging_app_subscribe_endpoint	(	const char *	app_name,
		struct ast_endpoint *	endpoint,
		message_received_cb	callback,
		void *	pvt
	)

Subscribe an application to an endpoint for messages.

Parameters

app_name	The name of the Stasis Message Bus API application to subscribe to endpoint
endpoint	The endpoint object to subscribe to
callback	The callback to call when a message is received
pvt	An ao2 ref counted object that will be passed to the callback.

Return values

0	subscription was successful
-1	subscription failed

Definition at line 493 of file messaging.c.

{
    RAII_VAR(struct message_subscription *, sub, NULL, ao2_cleanup);
    struct application_tuple *tuple;
 
    sub = get_or_create_subscription(endpoint);
    if (!sub) {
        return -1;
    }
 
    ao2_lock(sub);
    if (is_app_subscribed(sub, app_name)) {
        ao2_unlock(sub);
        return 0;
    }
 
    tuple = application_tuple_alloc(app_name, callback, pvt);
    if (!tuple) {
        ao2_unlock(sub);
        return -1;
    }
    if (AST_VECTOR_APPEND(&sub->applications, tuple)) {
        ao2_ref(tuple, -1);
        ao2_unlock(sub);
        return -1;
    }
    ao2_unlock(sub);
 
    ast_debug(3, "App '%s' subscribed to messages from endpoint '%s'\n", app_name, endpoint ? ast_endpoint_get_id(endpoint) : "-- ALL --");
    ast_test_suite_event_notify("StasisMessagingSubscription", "SubState: Subscribed\r\nAppName: %s\r\nToken: %s\r\n",
        app_name, endpoint ? ast_endpoint_get_id(endpoint) : "ALL");
 
    return 0;
}

References ao2_cleanup, ao2_lock, ao2_ref, ao2_unlock, app_name(), application_tuple_alloc(), ast_debug, ast_endpoint_get_id(), ast_test_suite_event_notify, AST_VECTOR_APPEND, callback(), get_or_create_subscription(), is_app_subscribed(), NULL, application_tuple::pvt, RAII_VAR, and sub.

Referenced by app_subscribe_endpoint().

messaging_app_unsubscribe_endpoint()

void messaging_app_unsubscribe_endpoint	(	const char *	app_name,
		const char *	endpoint_id
	)

Subscribe for messages from a particular endpoint.

Parameters

app_name	Name of the stasis application to unsubscribe from messaging
endpoint_id	The ID of the endpoint we no longer care about

Definition at line 423 of file messaging.c.

{
    RAII_VAR(struct message_subscription *, sub, NULL, ao2_cleanup);
    RAII_VAR(struct ast_endpoint *, endpoint, NULL, ao2_cleanup);
 
    endpoint = ast_endpoint_find_by_id(endpoint_id);
    sub = get_subscription(endpoint);
    if (!sub) {
        return;
    }
 
    ao2_lock(sub);
    if (!is_app_subscribed(sub, app_name)) {
        ao2_unlock(sub);
        return;
    }
 
    AST_VECTOR_REMOVE_CMP_UNORDERED(&sub->applications, app_name, application_tuple_cmp, ao2_cleanup);
    if (AST_VECTOR_SIZE(&sub->applications) == 0) {
        if (endpoint && !ast_strlen_zero(ast_endpoint_get_resource(endpoint))) {
            ao2_unlink(endpoint_subscriptions, sub);
        } else {
            ast_rwlock_wrlock(&tech_subscriptions_lock);
            AST_VECTOR_REMOVE_CMP_UNORDERED(&tech_subscriptions, endpoint ? ast_endpoint_get_id(endpoint) : TECH_WILDCARD,
                messaging_subscription_cmp, AST_VECTOR_ELEM_CLEANUP_NOOP);
            ast_rwlock_unlock(&tech_subscriptions_lock);
            ao2_ref(sub, -1); /* Release the reference held by tech_subscriptions */
        }
    }
    ao2_unlock(sub);
 
    ast_debug(3, "App '%s' unsubscribed to messages from endpoint '%s'\n", app_name, endpoint ? ast_endpoint_get_id(endpoint) : "-- ALL --");
    ast_test_suite_event_notify("StasisMessagingSubscription", "SubState: Unsubscribed\r\nAppName: %s\r\nToken: %s\r\n",
        app_name, endpoint ? ast_endpoint_get_id(endpoint) : "ALL");
}

References ao2_cleanup, ao2_lock, ao2_ref, ao2_unlink, ao2_unlock, app_name(), application_tuple_cmp(), ast_debug, ast_endpoint_find_by_id(), ast_endpoint_get_id(), ast_endpoint_get_resource(), ast_rwlock_unlock, ast_rwlock_wrlock, ast_strlen_zero(), ast_test_suite_event_notify, AST_VECTOR_ELEM_CLEANUP_NOOP, AST_VECTOR_REMOVE_CMP_UNORDERED, AST_VECTOR_SIZE, endpoint_subscriptions, get_subscription(), is_app_subscribed(), messaging_subscription_cmp(), NULL, RAII_VAR, sub, tech_subscriptions, tech_subscriptions_lock, and TECH_WILDCARD.

Referenced by unsubscribe().

messaging_cleanup()

int messaging_cleanup

(

void

)

Tidy up the messaging layer.

Return values

0	success
-1	failure

Definition at line 529 of file messaging.c.

{
    ast_msg_handler_unregister(&ari_msg_handler);
    ao2_ref(endpoint_subscriptions, -1);
    AST_VECTOR_FREE(&tech_subscriptions);
    ast_rwlock_destroy(&tech_subscriptions_lock);\
 
    return 0;
}

References ao2_ref, ari_msg_handler, ast_msg_handler_unregister(), ast_rwlock_destroy, AST_VECTOR_FREE, endpoint_subscriptions, tech_subscriptions, and tech_subscriptions_lock.

Referenced by unload_module().

messaging_init()

int messaging_init

(

void

)

Initialize the messaging layer.

Return values

0	success
-1	failure

Definition at line 539 of file messaging.c.

{
    endpoint_subscriptions = ao2_t_container_alloc_hash(AO2_ALLOC_OPT_LOCK_RWLOCK, 0,
        ENDPOINTS_NUM_BUCKETS, message_subscription_hash_cb, NULL,
        message_subscription_compare_cb, "Endpoint messaging subscription container creation");
    if (!endpoint_subscriptions) {
        return -1;
    }
 
    if (AST_VECTOR_INIT(&tech_subscriptions, 4)) {
        ao2_ref(endpoint_subscriptions, -1);
        return -1;
    }
 
    if (ast_rwlock_init(&tech_subscriptions_lock)) {
        ao2_ref(endpoint_subscriptions, -1);
        AST_VECTOR_FREE(&tech_subscriptions);
        return -1;
    }
 
    if (ast_msg_handler_register(&ari_msg_handler)) {
        ao2_ref(endpoint_subscriptions, -1);
        AST_VECTOR_FREE(&tech_subscriptions);
        ast_rwlock_destroy(&tech_subscriptions_lock);
        return -1;
    }
 
    return 0;
}

References AO2_ALLOC_OPT_LOCK_RWLOCK, ao2_ref, ao2_t_container_alloc_hash, ari_msg_handler, ast_msg_handler_register(), ast_rwlock_destroy, ast_rwlock_init, AST_VECTOR_FREE, AST_VECTOR_INIT, endpoint_subscriptions, ENDPOINTS_NUM_BUCKETS, message_subscription_compare_cb(), message_subscription_hash_cb(), NULL, tech_subscriptions, and tech_subscriptions_lock.

Referenced by load_module().

messaging_subscription_cmp()

static int messaging_subscription_cmp ( struct message_subscription *  sub, const char *  key  )

static

Definition at line 368 of file messaging.c.

{
    return !strcmp(sub->token, key) ? 1 : 0;
}

References sub.

Referenced by messaging_app_unsubscribe_endpoint().

msg_to_endpoint()

static void msg_to_endpoint ( const struct ast_msg *  msg, char *  buf, size_t  len  )

static

Definition at line 200 of file messaging.c.

{
    const char *endpoint = ast_msg_get_endpoint(msg);
 
    snprintf(buf, len, "%s%s%s", ast_msg_get_tech(msg),
        ast_strlen_zero(endpoint) ? "" : "/",
        S_OR(endpoint, ""));
}

References ast_msg_get_endpoint(), ast_msg_get_tech(), ast_strlen_zero(), buf, len(), and S_OR.

Referenced by handle_msg_cb(), and has_destination_cb().

msg_to_json()

static struct ast_json * msg_to_json ( struct ast_msg *  msg )

static

Definition at line 252 of file messaging.c.

{
    struct ast_json *json_obj;
    struct ast_json *json_vars;
    struct ast_msg_var_iterator *it_vars;
    const char *name;
    const char *value;
 
    it_vars = ast_msg_var_iterator_init(msg);
    if (!it_vars) {
        return NULL;
    }
 
    json_vars = ast_json_object_create();
    if (!json_vars) {
        ast_msg_var_iterator_destroy(it_vars);
        return NULL;
    }
 
    while (ast_msg_var_iterator_next_received(msg, it_vars, &name, &value)) {
        struct ast_json *json_val = ast_json_string_create(value);
        if (!json_val || ast_json_object_set(json_vars, name, json_val)) {
            ast_json_unref(json_vars);
            ast_msg_var_iterator_destroy(it_vars);
            return NULL;
        }
 
        ast_msg_var_unref_current(it_vars);
    }
    ast_msg_var_iterator_destroy(it_vars);
 
    json_obj = ast_json_pack("{s: s, s: s, s: s, s: o}",
        "from", ast_msg_get_from(msg),
        "to", ast_msg_get_to(msg),
        "body", ast_msg_get_body(msg),
        "variables", json_vars);
 
    return json_obj;
}

References ast_json_object_create(), ast_json_object_set(), ast_json_pack(), ast_json_string_create(), ast_json_unref(), ast_msg_get_body(), ast_msg_get_from(), ast_msg_get_to(), ast_msg_var_iterator_destroy(), ast_msg_var_iterator_init(), ast_msg_var_iterator_next_received(), ast_msg_var_unref_current(), name, NULL, and value.

Referenced by handle_msg_cb().

Variable Documentation

ari_msg_handler

struct ast_msg_handler ari_msg_handler

Initial value:

= {
    .name = "ari",
    .handle_msg = handle_msg_cb,
    .has_destination = has_destination_cb,
}

Definition at line 362 of file messaging.c.

                                         {
    .name = "ari",
    .handle_msg = handle_msg_cb,
    .has_destination = has_destination_cb,
};

Referenced by messaging_cleanup(), and messaging_init().

current

size_t current

Definition at line 75 of file messaging.c.

elems

struct message_subscription* * elems

Definition at line 75 of file messaging.c.

endpoint_subscriptions

struct ao2_container* endpoint_subscriptions

static

The subscriptions to endpoints.

Definition at line 67 of file messaging.c.

Referenced by get_or_create_subscription(), get_subscription(), handle_msg_cb(), has_destination_cb(), messaging_app_unsubscribe_endpoint(), messaging_cleanup(), and messaging_init().

max

size_t max

Definition at line 75 of file messaging.c.

[struct]

struct { ... } tech_subscriptions

The subscriptions to technologies.

Note

These are stored separately from standard endpoints, given how relatively few of them there are.

Referenced by get_or_create_subscription(), get_subscription(), handle_msg_cb(), has_destination_cb(), messaging_app_unsubscribe_endpoint(), messaging_cleanup(), and messaging_init().

tech_subscriptions_lock

ast_rwlock_t tech_subscriptions_lock

static

RWLock for tech_subscriptions.

Definition at line 78 of file messaging.c.

Referenced by get_or_create_subscription(), get_subscription(), handle_msg_cb(), has_destination_cb(), messaging_app_unsubscribe_endpoint(), messaging_cleanup(), and messaging_init().