stasis_internal.h File Reference

Internal Stasis APIs. More...

#include "asterisk/stasis.h"

Include dependency graph for stasis_internal.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Functions
struct stasis_subscription *	internal_stasis_subscribe (struct stasis_topic *topic, stasis_subscription_cb callback, void *data, int needs_mailbox, int use_thread_pool, const char *file, int lineno, const char *func)
	Create a subscription. More...

Detailed Description

Internal Stasis APIs.

This header file is used to define functions that are shared between files that make up Stasis Message Bus API. Functions declared here should not be used by any module outside of Stasis.

If you find yourself needing to call one of these functions directly, something has probably gone horribly wrong.

Author

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

Definition in file stasis_internal.h.

Function Documentation

internal_stasis_subscribe()

struct stasis_subscription * internal_stasis_subscribe	(	struct stasis_topic *	topic,
		stasis_subscription_cb	callback,
		void *	data,
		int	needs_mailbox,
		int	use_thread_pool,
		const char *	file,
		int	lineno,
		const char *	func
	)

Create a subscription.

In addition to being AO2 managed memory (requiring an ao2_cleanup() to free up this reference), the subscription must be explicitly unsubscribed from its topic using stasis_unsubscribe().

The invocations of the callback are serialized, but may not always occur on the same thread. The invocation order of different subscriptions is unspecified.

Note: modules outside of Stasis should use stasis_subscribe.

Parameters

topic	Topic to subscribe to.
callback	Callback function for subscription messages.
data	Data to be passed to the callback, in addition to the message.
needs_mailbox	Determines whether or not the subscription requires a mailbox. Subscriptions with mailboxes will be delivered on some non-publisher thread; subscriptions without mailboxes will be delivered on the publisher thread.
use_thread_pool	Use the thread pool for the subscription. This is only relevant if needs_mailbox is non-zero.
file,lineno,func

Returns

New stasis_subscription object.

Return values

NULL

on error.

Since

12

Definition at line 923 of file stasis.c.

{
    struct stasis_subscription *sub;
    int ret;
 
    if (!topic) {
        return NULL;
    }
 
    /* The ao2 lock is used for join_cond. */
    sub = ao2_t_alloc(sizeof(*sub), subscription_dtor, stasis_topic_name(topic));
    if (!sub) {
        return NULL;
    }
 
#ifdef AST_DEVMODE
    ret = ast_asprintf(&sub->uniqueid, "%s:%s-%d", file, stasis_topic_name(topic), ast_atomic_fetchadd_int(&topic->subscriber_id, +1));
    sub->statistics = stasis_subscription_statistics_create(sub, needs_mailbox, use_taskpool, file, lineno, func);
    if (ret < 0 || !sub->statistics) {
        ao2_ref(sub, -1);
        return NULL;
    }
#else
    ret = ast_asprintf(&sub->uniqueid, "%s-%d", stasis_topic_name(topic), ast_atomic_fetchadd_int(&topic->subscriber_id, +1));
    if (ret < 0) {
        ao2_ref(sub, -1);
        return NULL;
    }
#endif
 
    if (needs_mailbox) {
        char tps_name[AST_TASKPROCESSOR_MAX_NAME + 1];
 
        /* Create name with seq number appended. */
        ast_taskprocessor_build_name(tps_name, sizeof(tps_name), "stasis/%c:%s",
            use_taskpool ? 'p' : 'm',
            stasis_topic_name(topic));
 
        /*
         * With a small number of subscribers, a thread-per-sub is
         * acceptable. For a large number of subscribers, a thread
         * pool should be used.
         */
        if (use_taskpool) {
            sub->mailbox = ast_taskpool_serializer(tps_name, taskpool);
        } else {
            sub->mailbox = ast_taskprocessor_get(tps_name, TPS_REF_DEFAULT);
        }
        if (!sub->mailbox) {
            ao2_ref(sub, -1);
 
            return NULL;
        }
        ast_taskprocessor_set_local(sub->mailbox, sub);
        /* Taskprocessor has a reference */
        ao2_ref(sub, +1);
    }
 
    ao2_ref(topic, +1);
    sub->topic = topic;
    sub->callback = callback;
    sub->data = data;
    ast_cond_init(&sub->join_cond, NULL);
    sub->filter = STASIS_SUBSCRIPTION_FILTER_NONE;
    AST_VECTOR_INIT(&sub->accepted_message_types, 0);
    sub->accepted_formatters = STASIS_SUBSCRIPTION_FORMATTER_NONE;
 
    if (topic_add_subscription(topic, sub) != 0) {
        ao2_ref(sub, -1);
        ao2_ref(topic, -1);
 
        return NULL;
    }
    send_subscription_subscribe(topic, sub);
 
    return sub;
}

References ao2_ref, ao2_t_alloc, ast_asprintf, ast_atomic_fetchadd_int(), ast_cond_init, ast_taskpool_serializer(), ast_taskprocessor_build_name(), ast_taskprocessor_get(), AST_TASKPROCESSOR_MAX_NAME, ast_taskprocessor_set_local(), AST_VECTOR_INIT, callback(), stasis_subscription::data, make_ari_stubs::file, NULL, send_subscription_subscribe(), STASIS_SUBSCRIPTION_FILTER_NONE, STASIS_SUBSCRIPTION_FORMATTER_NONE, stasis_topic_name(), statistics(), sub, stasis_topic::subscriber_id, subscription_dtor(), taskpool, stasis_subscription::topic, topic_add_subscription(), and TPS_REF_DEFAULT.

Referenced by __stasis_subscribe(), __stasis_subscribe_pool(), and stasis_caching_topic_create().