stasis_app_broadcast.h File Reference

Stasis Application Broadcast API. More...

#include "asterisk/channel.h"
#include "asterisk/optional_api.h"

Include dependency graph for stasis_app_broadcast.h:

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

Go to the source code of this file.

Macros
#define	STASIS_BROADCAST_FLAG_SUPPRESS_CLAIMED   (1 << 0)
	Suppress CallClaimed event for this broadcast.

Functions
int AST_OPTIONAL_API_NAME()	stasis_app_broadcast_channel (struct ast_channel *chan, int timeout_ms, const char *app_filter, unsigned int flags)
	Start a broadcast for a channel.
void AST_OPTIONAL_API_NAME()	stasis_app_broadcast_cleanup (const char *channel_id)
	Clean up broadcast context for a channel.
int AST_OPTIONAL_API_NAME()	stasis_app_broadcast_wait (struct ast_channel *chan, int timeout_ms)
	Wait for a broadcast channel to be claimed.
char *AST_OPTIONAL_API_NAME()	stasis_app_broadcast_winner (const char *channel_id)
	Get the winner app name for a broadcast channel.
int AST_OPTIONAL_API_NAME()	stasis_app_claim_channel (const char *channel_id, const char *app_name)
	Attempt to claim a broadcast channel.

Detailed Description

Stasis Application Broadcast API.

Author

Daniel Donoghue danie.nosp@m.l.do.nosp@m.noghu.nosp@m.e@au.nosp@m.rorai.nosp@m.nnov.nosp@m.ation.nosp@m..com

This module provides the infrastructure for broadcasting incoming channels to multiple ARI applications and handling first-claim winner logic.

Definition in file stasis_app_broadcast.h.

Macro Definition Documentation

STASIS_BROADCAST_FLAG_SUPPRESS_CLAIMED

#define STASIS_BROADCAST_FLAG_SUPPRESS_CLAIMED   (1 << 0)

Suppress CallClaimed event for this broadcast.

Definition at line 36 of file stasis_app_broadcast.h.

Function Documentation

stasis_app_broadcast_channel()

int AST_OPTIONAL_API_NAME() stasis_app_broadcast_channel	(	struct ast_channel *	chan,
		int	timeout_ms,
		const char *	app_filter,
		unsigned int	flags
	)

Start a broadcast for a channel.

Since

20

Broadcasts a channel to all ARI applications (or filtered applications) allowing them to claim the channel. Only the first claim will succeed.

When a channel is claimed, a CallClaimed event is sent only to applications that matched the app_filter (or all apps if no filter was set). This can be suppressed entirely with STASIS_BROADCAST_FLAG_SUPPRESS_CLAIMED.

Parameters

chan	The channel to broadcast
timeout_ms	Timeout in milliseconds to wait for a claim
app_filter	Optional regex filter for application names (NULL for all)
flags	Combination of STASIS_BROADCAST_FLAG_* values

Return values

0	on success
-1	on error
AST_OPTIONAL_API_UNAVAILABLE	if res_stasis_broadcast is not loaded

Parameters

chan	The channel to broadcast
timeout_ms	Timeout in milliseconds
app_filter	Optional regex filter for applications

Returns

0 on success, -1 on error

Definition at line 550 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
    struct ast_datastore *datastore;
 
    if (!chan) {
        return -1;
    }
 
    if (!broadcast_contexts) {
        return -1;
    }
 
    /* Remove any previous broadcast datastore from a prior attempt.
     * This supports failover scenarios where StasisBroadcast() is
     * called multiple times for the same channel.  The datastore
     * destructor unlinks the old context from the container. */
    {
        struct ast_datastore *old_ds;
        ast_channel_lock(chan);
        old_ds = ast_channel_datastore_find(chan, &broadcast_datastore_info, NULL);
        if (old_ds) {
            ast_channel_datastore_remove(chan, old_ds);
        }
        ast_channel_unlock(chan);
        if (old_ds) {
            ast_datastore_free(old_ds);
        }
    }
 
    /* Create broadcast context (validates and compiles app_filter regex) */
    ctx = broadcast_ctx_create(ast_channel_uniqueid(chan), app_filter, flags);
    if (!ctx) {
        ast_log(LOG_ERROR, "Channel %s: failed to create broadcast context\n",
            ast_channel_uniqueid(chan));
        return -1;
    }
 
    /* Store context in container */
    ao2_link(broadcast_contexts, ctx);
 
    /* Create and attach datastore to channel */
    datastore = ast_datastore_alloc(&broadcast_datastore_info, ast_channel_uniqueid(chan));
    if (!datastore) {
        ast_log(LOG_ERROR, "Channel %s: failed to allocate broadcast datastore\n",
            ast_channel_uniqueid(chan));
        ao2_unlink(broadcast_contexts, ctx);
        return -1;
    }
 
    datastore->data = ao2_bump(ctx);
    ast_channel_lock(chan);
    if (ast_channel_datastore_add(chan, datastore)) {
        ast_channel_unlock(chan);
        ast_log(LOG_ERROR, "Channel %s: failed to attach broadcast datastore\n",
            ast_channel_uniqueid(chan));
        ast_datastore_free(datastore);
        ao2_unlink(broadcast_contexts, ctx);
        return -1;
    }
    ast_channel_unlock(chan);
 
    ast_debug(1, "Starting broadcast for channel %s (timeout: %dms, filter: %s)\n",
        ast_channel_uniqueid(chan), timeout_ms, app_filter ? app_filter : "none");
 
    /* Send broadcast event to all matching applications */
    if (send_broadcast_event(chan, ctx) != 0) {
        ast_log(LOG_ERROR, "Channel %s: failed to send broadcast event\n",
            ast_channel_uniqueid(chan));
        ast_channel_lock(chan);
        ast_channel_datastore_remove(chan, datastore);
        ast_channel_unlock(chan);
        ast_datastore_free(datastore);
        ao2_unlink(broadcast_contexts, ctx);
        return -1;
    }
 
    return 0;
}

References ao2_bump, ao2_cleanup, ao2_link, ao2_unlink, ast_channel_datastore_add(), ast_channel_datastore_find(), ast_channel_datastore_remove(), ast_channel_lock, ast_channel_uniqueid(), ast_channel_unlock, ast_datastore_alloc, ast_datastore_free(), ast_debug, ast_log, broadcast_contexts, broadcast_ctx_create(), broadcast_datastore_info, ast_datastore::data, LOG_ERROR, NULL, RAII_VAR, and send_broadcast_event().

Referenced by stasis_broadcast_exec().

stasis_app_broadcast_cleanup()

void AST_OPTIONAL_API_NAME() stasis_app_broadcast_cleanup

(

const char *

channel_id

)

Clean up broadcast context for a channel.

Since

20

Removes the broadcast context when the channel is done or leaving the broadcast state.

Parameters

channel_id

The unique ID of the channel

This is the normal-path cleanup called by the dialplan application after the broadcast completes. The channel datastore destructor (broadcast_datastore_destroy) also unlinks the context as a safety net for abnormal teardown; ao2_unlink is idempotent so the double call is harmless.

Parameters

channel_id

The unique ID of the channel

Definition at line 885 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
 
    if (ast_strlen_zero(channel_id) || !broadcast_contexts) {
        return;
    }
 
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY | OBJ_UNLINK);
    if (ctx) {
        ast_debug(3, "Cleaning up broadcast context for %s\n", channel_id);
    }
}

References ao2_cleanup, ao2_find, ast_debug, ast_strlen_zero(), broadcast_contexts, NULL, OBJ_SEARCH_KEY, OBJ_UNLINK, and RAII_VAR.

Referenced by stasis_broadcast_exec().

stasis_app_broadcast_wait()

int AST_OPTIONAL_API_NAME() stasis_app_broadcast_wait	(	struct ast_channel *	chan,
		int	timeout_ms
	)

Wait for a broadcast channel to be claimed.

Since

20

Blocks until the channel is claimed or the timeout expires.

Parameters

chan	The channel
timeout_ms	Maximum time to wait in milliseconds

Return values

0	if claimed within timeout
-1	if timeout expired or error
AST_OPTIONAL_API_UNAVAILABLE	if res_stasis_broadcast is not loaded

Blocks until the channel is claimed, the timeout expires, or the channel hangs up. The hangup check runs every BROADCAST_POLL_INTERVAL_MS so that a dead channel does not tie up a PBX thread for the full timeout period.

Parameters

chan	The channel
timeout_ms	Maximum time to wait in milliseconds

Returns

0 if claimed within timeout, -1 otherwise

Definition at line 782 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
    const char *channel_id;
    struct timeval deadline;
    int result = -1;
 
    if (!chan) {
        return -1;
    }
 
    channel_id = ast_channel_uniqueid(chan);
 
    if (!broadcast_contexts) {
        return -1;
    }
 
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY);
    if (!ctx) {
        ast_log(LOG_WARNING, "No broadcast context for channel %s\n", channel_id);
        return -1;
    }
 
    /* Cap excessive timeouts to prevent arithmetic overflow */
    if (timeout_ms < 0) {
        timeout_ms = 0;
    } else if (timeout_ms > MAX_BROADCAST_TIMEOUT_MS) {
        timeout_ms = MAX_BROADCAST_TIMEOUT_MS;
    }
 
    /* Calculate absolute deadline */
    deadline = ast_tvadd(ast_tvnow(),
        ast_tv(timeout_ms / 1000, (timeout_ms % 1000) * 1000));
 
    ao2_lock(ctx);
    while (!ctx->claimed) {
        struct timeval now;
        struct timespec poll_spec;
        long remaining_ms;
        long poll_ms;
        int wait_result;
 
        /* Check for hangup so we don't block on a dead channel */
        if (ast_check_hangup(chan)) {
            ast_debug(3, "Channel %s hung up during broadcast wait\n",
                channel_id);
            break;
        }
 
        /* Check if we've passed the overall deadline */
        now = ast_tvnow();
        remaining_ms = ast_tvdiff_ms(deadline, now);
        if (remaining_ms <= 0) {
            ast_debug(3, "Broadcast timeout for channel %s after %dms\n",
                channel_id, timeout_ms);
            break;
        }
 
        /* Sleep for the shorter of the remaining time and the poll interval */
        poll_ms = remaining_ms;
        if (poll_ms > BROADCAST_POLL_INTERVAL_MS) {
            poll_ms = BROADCAST_POLL_INTERVAL_MS;
        }
 
        poll_spec.tv_sec = now.tv_sec + (poll_ms / 1000);
        poll_spec.tv_nsec = (long)(now.tv_usec) * 1000L
            + (long)(poll_ms % 1000) * 1000000L;
        while (poll_spec.tv_nsec >= 1000000000) {
            poll_spec.tv_sec++;
            poll_spec.tv_nsec -= 1000000000;
        }
 
        wait_result = ast_cond_timedwait(&ctx->cond, ao2_object_get_lockaddr(ctx), &poll_spec);
        if (wait_result != 0 && wait_result != ETIMEDOUT) {
            ast_log(LOG_WARNING,
                "Channel %s: unexpected error waiting for claim: %s (%d)\n",
                channel_id, strerror(wait_result), wait_result);
            break;
        }
        /* Loop back: re-check claimed, then hangup, then deadline */
    }
 
    if (ctx->claimed) {
        ast_debug(1, "Channel %s claimed by %s\n",
            channel_id, ctx->winner_app);
        result = 0;
    }
    ao2_unlock(ctx);
 
    return result;
}

References ao2_cleanup, ao2_find, ao2_lock, ao2_object_get_lockaddr(), ao2_unlock, ast_channel_uniqueid(), ast_check_hangup(), ast_cond_timedwait, ast_debug, ast_log, ast_tv(), ast_tvadd(), ast_tvdiff_ms(), ast_tvnow(), broadcast_contexts, BROADCAST_POLL_INTERVAL_MS, LOG_WARNING, MAX_BROADCAST_TIMEOUT_MS, NULL, OBJ_SEARCH_KEY, RAII_VAR, result, and wait_result().

Referenced by stasis_broadcast_exec().

stasis_app_broadcast_winner()

char *AST_OPTIONAL_API_NAME() stasis_app_broadcast_winner

(

const char *

channel_id

)

Get the winner app name for a broadcast channel.

Since

20

Parameters

channel_id

The unique ID of the channel

Returns

A copy of the winner app name (caller must free with ast_free), or NULL if not claimed or not found

Return values

NULL	if res_stasis_broadcast is not loaded

Parameters

channel_id

The unique ID of the channel

Returns

A copy of the winner app name (caller must free with ast_free), or NULL if not claimed or not found

Definition at line 743 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
    char *winner = NULL;
 
    if (ast_strlen_zero(channel_id)) {
        return NULL;
    }
 
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY);
    if (!ctx) {
        return NULL;
    }
 
    ao2_lock(ctx);
    if (ctx->claimed) {
        winner = ast_strdup(ctx->winner_app);
    }
    /* Mark the broadcast as finished so no new claims can succeed.
     * This closes the race window between reading the winner and
     * the subsequent broadcast_cleanup call. */
    ctx->finished = 1;
    ao2_unlock(ctx);
 
    return winner;
}

References ao2_cleanup, ao2_find, ao2_lock, ao2_unlock, ast_strdup, ast_strlen_zero(), broadcast_contexts, NULL, OBJ_SEARCH_KEY, and RAII_VAR.

Referenced by stasis_broadcast_exec().

stasis_app_claim_channel()

int AST_OPTIONAL_API_NAME() stasis_app_claim_channel	(	const char *	channel_id,
		const char *	app_name
	)

Attempt to claim a broadcast channel.

Since

20

Atomically attempts to claim a channel that is in broadcast state. Only the first claim for a given channel will succeed.

Parameters

channel_id	The unique ID of the channel
app_name	The name of the application claiming the channel

Return values

0	if claim successful
-1	if channel not found
-2	if already claimed by another application
AST_OPTIONAL_API_UNAVAILABLE	if res_stasis_broadcast is not loaded

Parameters

channel_id	The unique ID of the channel
app_name	The name of the application claiming the channel

Returns

0 if claim successful, -1 if channel not found, -2 if already claimed

Definition at line 637 of file res_stasis_broadcast.c.

{
    RAII_VAR(struct stasis_broadcast_ctx *, ctx, NULL, ao2_cleanup);
 
    if (ast_strlen_zero(channel_id) || ast_strlen_zero(app_name)) {
        return -1;
    }
 
    if (!broadcast_contexts) {
        return -1;
    }
 
    /* Find broadcast context */
    ctx = ao2_find(broadcast_contexts, channel_id, OBJ_SEARCH_KEY);
    if (!ctx) {
        ast_debug(1, "No broadcast context found for channel %s\n", channel_id);
        return -1;
    }
 
    /* Atomically check and set claimed flag.
     * Check claimed before finished: if the channel was claimed and then the
     * broadcast finished, a late claim should return -2 (409 Conflict) rather
     * than -1 (404) so callers can distinguish "already taken" from "not found". */
    ao2_lock(ctx);
    if (ctx->claimed) {
        ast_debug(1, "Channel %s already claimed by %s (attempt by %s denied)\n",
            channel_id, ctx->winner_app ? ctx->winner_app : "(unknown)", app_name);
        ao2_unlock(ctx);
        return -2;
    }
    if (ctx->finished) {
        ast_debug(1, "Channel %s broadcast already finished (late claim by %s rejected)\n",
            channel_id, app_name);
        ao2_unlock(ctx);
        return -1;
    }
    ctx->winner_app = ast_strdup(app_name);
    if (!ctx->winner_app) {
        ast_log(LOG_ERROR,
            "Failed to allocate winner app name for channel %s\n",
            channel_id);
        ao2_unlock(ctx);
        return -1;
    }
    ctx->claimed = 1;
    ast_verb(3, "Channel %s claimed by application %s\n",
        channel_id, app_name);
    /* Signal waiting thread that channel was claimed */
    ast_cond_signal(&ctx->cond);
    ao2_unlock(ctx);
 
    /* Send CallClaimed event to matching apps */
    if (!(ctx->flags & STASIS_BROADCAST_FLAG_SUPPRESS_CLAIMED)) {
        RAII_VAR(struct ast_channel_snapshot *, snapshot, NULL, ao2_cleanup);
        RAII_VAR(struct ast_json *, event, NULL, ast_json_unref);
        RAII_VAR(struct ao2_container *, apps, NULL, ao2_cleanup);
        struct ao2_iterator iter;
        char *app_name_iter;
 
        snapshot = ast_channel_snapshot_get_latest(channel_id);
        if (snapshot) {
            event = ast_json_pack("{s: s, s: o, s: o, s: s}",
                "type", "CallClaimed",
                "timestamp", ast_json_timeval(ast_tvnow(), NULL),
                "channel", ast_channel_snapshot_to_json(snapshot, NULL),
                "winner_app", app_name);
        }
        if (event) {
            apps = stasis_app_get_all();
        }
        if (apps) {
            iter = ao2_iterator_init(apps, 0);
            while ((app_name_iter = ao2_iterator_next(&iter))) {
                struct ast_json *event_copy;
 
                /* Only send to apps that matched the original broadcast filter */
                if (ctx->filter_compiled &&
                    regexec(&ctx->compiled_filter, app_name_iter,
                        0, NULL, 0) == REG_NOMATCH) {
                    ao2_ref(app_name_iter, -1);
                    continue;
                }
 
                event_copy = ast_json_deep_copy(event);
                if (!event_copy) {
                    ao2_ref(app_name_iter, -1);
                    continue;
                }
 
                stasis_app_send(app_name_iter, event_copy);
                ast_json_unref(event_copy);
                ao2_ref(app_name_iter, -1);
            }
            ao2_iterator_destroy(&iter);
        }
    }
 
    return 0;
}

References ao2_cleanup, ao2_find, ao2_iterator_destroy(), ao2_iterator_init(), ao2_iterator_next, ao2_lock, ao2_ref, ao2_unlock, app_name(), ast_channel_snapshot_get_latest(), ast_channel_snapshot_to_json(), ast_cond_signal, ast_debug, ast_json_deep_copy(), ast_json_pack(), ast_json_timeval(), ast_json_unref(), ast_log, ast_strdup, ast_strlen_zero(), ast_tvnow(), ast_verb, broadcast_contexts, LOG_ERROR, NULL, OBJ_SEARCH_KEY, RAII_VAR, stasis_app_get_all(), stasis_app_send(), and STASIS_BROADCAST_FLAG_SUPPRESS_CLAIMED.

Referenced by ast_ari_events_claim_channel().