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:

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

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_task_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_task_pool	Use the task 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;
}

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