Asterisk
/
asterisk
mirror of https://github.com/asterisk/asterisk
1
0
Fork 0
Code Issues Releases Wiki Activity

Remove required type field from channel blobs

Browse Source 
When we first introduced the channel blob types, the JSON blobs were
self identifying by a required "type" field in the JSON object
itself. This, as it turns out, was a bad idea.

When we introduced the message router, it was useless for routing based
on the JSON type. And messages had two type fields to check: the
stasis_message_type() of the message itself, plus the type field in the
JSON blob (but only if it was a blob message).

This patch corrects that mistake by removing the required type field
from JSON blobs, and introducing first class stasis_message_type objects
for the actual message type.

Since we now will have a proliferation of message types, I introduced a
few macros to help reduce the amount of boilerplate necessary to set
them up.

Review: https://reviewboard.asterisk.org/r/2509


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@388005 65c4cc65-6c06-0410-ace0-fbb531ad65f3
changes/78/78/1
David M. Lee 12 years ago
parent 297feffd4e
commit 0eb4cf8c19
8 changed files with 238 additions and 201 deletions
Show Stats Download Patch File Download Diff File

4
apps/app_userevent.c
Unescape View File

@ -93,7 +93,6 @@ static int userevent_exec(struct ast_channel *chan, const char *data)
} }
blob = ast_json_pack("{s: s, s: s, s: s}", blob = ast_json_pack("{s: s, s: s, s: s}",
"type", "userevent",
"eventname", args.eventname, "eventname", args.eventname,
"body", ast_str_buffer(body)); "body", ast_str_buffer(body));
if (!blob) { if (!blob) {
@ -101,7 +100,8 @@ static int userevent_exec(struct ast_channel *chan, const char *data)
return -1; return -1;
} }
msg = ast_channel_blob_create(chan, blob); msg = ast_channel_blob_create(
chan, ast_channel_user_event_type(), blob);
if (!msg) { if (!msg) {
return -1; return -1;
} }

37
include/asterisk/stasis.h
Unescape View File

@ -542,6 +542,43 @@ struct ao2_container *stasis_cache_dump(struct stasis_caching_topic *caching_top
/*! @{ */ /*! @{ */
/*!
* \brief Boiler-plate removing macro for defining message types.
*
* \param name Name of message type.
* \since 12
*/
#define STASIS_MESSAGE_TYPE_DEFN(name) \
static struct stasis_message_type *__ ## name; \
struct stasis_message_type *name(void) { \
ast_assert(__ ## name != NULL); \
return __ ## name; \
}
/*!
* \brief Boiler-plate removing macro for initializing message types.
*
* \param name Name of message type.
* \return 0 if initialization is successful.
* \return Non-zero on failure.
* \since 12
*/
#define STASIS_MESSAGE_TYPE_INIT(name) \
({ \
__ ## name = stasis_message_type_create(#name); \
__ ## name ? 0 : -1; \
})
#define STASIS_MESSAGE_TYPE_CLEANUP(name) \
({ \
ao2_cleanup(__ ## name); \
__ ## name = NULL; \
})
/*! @} */
/*! @{ */
/*! /*!
* \brief Initialize the Stasis subsystem * \brief Initialize the Stasis subsystem
* \return 0 on success. * \return 0 on success.

108
include/asterisk/stasis_channels.h
Unescape View File

@ -70,8 +70,7 @@ struct ast_channel_snapshot {
* \since 12 * \since 12
* \brief Blob of data associated with a channel. * \brief Blob of data associated with a channel.
* *
* The \c blob is actually a JSON object of structured data. It has a "type" field * This blob is actually shared amongst several \ref stasis_message_type's.
* which contains the type string describing this blob.
*/ */
struct ast_channel_blob { struct ast_channel_blob {
/*! Channel blob is associated with (or NULL for global/all channels) */ /*! Channel blob is associated with (or NULL for global/all channels) */
@ -110,14 +109,6 @@ struct stasis_caching_topic *ast_channel_topic_all_cached(void);
*/ */
struct stasis_message_type *ast_channel_snapshot_type(void); struct stasis_message_type *ast_channel_snapshot_type(void);
/*!
* \since 12
* \brief Message type for \ref ast_channel_blob messages.
*
* \retval Message type for \ref ast_channel_blob messages.
*/
struct stasis_message_type *ast_channel_blob_type(void);
/*! /*!
* \since 12 * \since 12
* \brief Generate a snapshot of the channel state. This is an ao2 object, so * \brief Generate a snapshot of the channel state. This is an ao2 object, so
@ -128,44 +119,35 @@ struct stasis_message_type *ast_channel_blob_type(void);
* \retval pointer on success (must be ast_freed) * \retval pointer on success (must be ast_freed)
* \retval NULL on error * \retval NULL on error
*/ */
struct ast_channel_snapshot *ast_channel_snapshot_create(struct ast_channel *chan); struct ast_channel_snapshot *ast_channel_snapshot_create(
struct ast_channel *chan);
/*! /*!
* \since 12 * \since 12
* \brief Creates a \ref ast_channel_blob message. * \brief Creates a \ref ast_channel_blob message.
* *
* The \a blob JSON object requires a \c "type" field describing the blob. It * The given \a blob should be treated as immutable and not modified after it is
* should also be treated as immutable and not modified after it is put into the * put into the message.
* message. *
* \param chan Channel blob is associated with, or \c NULL for global/all channels.
* \param type Message type for this blob.
* \param blob JSON object representing the data, or \c NULL for no data. If
* \c NULL, ast_json_null() is put into the object.
* *
* \param chan Channel blob is associated with, or NULL for global/all channels.
* \param blob JSON object representing the data.
* \return \ref ast_channel_blob message. * \return \ref ast_channel_blob message.
* \return \c NULL on error * \return \c NULL on error
*/ */
struct stasis_message *ast_channel_blob_create(struct ast_channel *chan, struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
struct ast_json *blob); struct stasis_message_type *type, struct ast_json *blob);
/*!
* \since 12
* \brief Extracts the type field from a \ref ast_channel_blob.
* Returned \c char* is still owned by \a obj
* \param obj Channel blob object.
* \return Type field value from the blob.
* \return \c NULL on error.
*/
const char *ast_channel_blob_json_type(struct ast_channel_blob *obj);
/*! /*!
* \since 12 * \since 12
* \brief Create a \ref ast_multi_channel_blob suitable for a \ref stasis_message * \brief Create a \ref ast_multi_channel_blob suitable for a \ref stasis_message.
* *
* \note Similar to a \ref ast_channel_blob, the \ref ast_multi_channel_blob requires * The given \a blob should be treated as immutable and not modified after it is
* a \a blob JSON object containing a \c "type" field describing the blob. It * put into the message.
* should also be treated as immutable and not modified after it is put into the
* message.
* *
* \param blob The JSON blob that defines the type of this \ref ast_multi_channel_blob * \param blob The JSON blob that defines the data of this \ref ast_multi_channel_blob
* *
* \return \ref ast_multi_channel_blob object * \return \ref ast_multi_channel_blob object
* \return \c NULL on error * \return \c NULL on error
@ -189,8 +171,7 @@ struct ast_multi_channel_blob *ast_multi_channel_blob_create(struct ast_json *bl
* \retval NULL on error or not found for the role specified * \retval NULL on error or not found for the role specified
*/ */
struct ast_channel_snapshot *ast_multi_channel_blob_get_channel( struct ast_channel_snapshot *ast_multi_channel_blob_get_channel(
struct ast_multi_channel_blob *obj, struct ast_multi_channel_blob *obj, const char *role);
const char *role);
/*! /*!
* \since 12 * \since 12
@ -212,8 +193,7 @@ struct ast_channel_snapshot *ast_multi_channel_blob_get_channel(
* \retval NULL on error or not found for the role specified * \retval NULL on error or not found for the role specified
*/ */
struct ao2_container *ast_multi_channel_blob_get_channels( struct ao2_container *ast_multi_channel_blob_get_channels(
struct ast_multi_channel_blob *obj, struct ast_multi_channel_blob *obj, const char *role);
const char *role);
/*! /*!
* \since 12 * \since 12
@ -226,17 +206,6 @@ struct ao2_container *ast_multi_channel_blob_get_channels(
*/ */
struct ast_json *ast_multi_channel_blob_get_json(struct ast_multi_channel_blob *obj); struct ast_json *ast_multi_channel_blob_get_json(struct ast_multi_channel_blob *obj);
/*!
* \since 12
* \brief Extracts the type field from a \ref ast_multi_channel_blob.
* Returned \c char* is still owned by \a obj
*
* \param obj Channel blob object.
* \return Type field value from the blob.
* \return \c NULL on error.
*/
const char *ast_multi_channel_blob_get_type(struct ast_multi_channel_blob *obj);
/*! /*!
* \since 12 * \since 12
* \brief Add a \ref ast_channel_snapshot to a \ref ast_multi_channel_blob object * \brief Add a \ref ast_channel_snapshot to a \ref ast_multi_channel_blob object
@ -250,8 +219,7 @@ const char *ast_multi_channel_blob_get_type(struct ast_multi_channel_blob *obj);
* \ref ast_multi_channel_blob object * \ref ast_multi_channel_blob object
*/ */
void ast_multi_channel_blob_add_channel(struct ast_multi_channel_blob *obj, void ast_multi_channel_blob_add_channel(struct ast_multi_channel_blob *obj,
const char *role, const char *role, struct ast_channel_snapshot *snapshot);
struct ast_channel_snapshot *snapshot);
/*! /*!
* \since 12 * \since 12
@ -272,6 +240,46 @@ void ast_channel_publish_varset(struct ast_channel *chan,
*/ */
struct stasis_message_type *ast_channel_dial_type(void); struct stasis_message_type *ast_channel_dial_type(void);
/*!
* \since 12
* \brief Message type for when a variable is set on a channel.
*
* \retval A stasis message type
*/
struct stasis_message_type *ast_channel_varset_type(void);
/*!
* \since 12
* \brief Message type for when a custom user event is sent on a channel.
*
* \retval A stasis message type
*/
struct stasis_message_type *ast_channel_user_event_type(void);
/*!
* \since 12
* \brief Message type for when a hangup is requested on a channel.
*
* \retval A stasis message type
*/
struct stasis_message_type *ast_channel_hangup_request_type(void);
/*!
* \since 12
* \brief Message type for when DTMF begins on a channel.
*
* \retval A stasis message type
*/
struct stasis_message_type *ast_channel_dtmf_begin_type(void);
/*!
* \since 12
* \brief Message type for when DTMF ends on a channel.
*
* \retval A stasis message type
*/
struct stasis_message_type *ast_channel_dtmf_end_type(void);
/*! /*!
* \since 12 * \since 12
* \brief Publish in the \ref ast_channel_topic or \ref ast_channel_topic_all * \brief Publish in the \ref ast_channel_topic or \ref ast_channel_topic_all

29
main/channel.c
Unescape View File

@ -1368,11 +1368,12 @@ int ast_queue_frame_head(struct ast_channel *chan, struct ast_frame *fin)
} }
/*! \internal \brief Publish a channel blob message */ /*! \internal \brief Publish a channel blob message */
static void publish_channel_blob(struct ast_channel *chan, struct ast_json *blob) static void publish_channel_blob(struct ast_channel *chan,
struct stasis_message_type *type, struct ast_json *blob)
{ {
RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup); RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
if (blob) { if (blob) {
message = ast_channel_blob_create(chan, blob); message = ast_channel_blob_create(chan, type, blob);
} }
if (message) { if (message) {
stasis_publish(ast_channel_topic(chan), message); stasis_publish(ast_channel_topic(chan), message);
@ -1382,7 +1383,6 @@ static void publish_channel_blob(struct ast_channel *chan, struct ast_json *blob
/*! \brief Queue a hangup frame for channel */ /*! \brief Queue a hangup frame for channel */
int ast_queue_hangup(struct ast_channel *chan) int ast_queue_hangup(struct ast_channel *chan)
{ {
RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup); RAII_VAR(struct stasis_message *, message, NULL, ao2_cleanup);
struct ast_frame f = { AST_FRAME_CONTROL, .subclass.integer = AST_CONTROL_HANGUP }; struct ast_frame f = { AST_FRAME_CONTROL, .subclass.integer = AST_CONTROL_HANGUP };
int res; int res;
@ -1390,8 +1390,7 @@ int ast_queue_hangup(struct ast_channel *chan)
/* Yeah, let's not change a lock-critical value without locking */ /* Yeah, let's not change a lock-critical value without locking */
ast_channel_lock(chan); ast_channel_lock(chan);
ast_channel_softhangup_internal_flag_add(chan, AST_SOFTHANGUP_DEV); ast_channel_softhangup_internal_flag_add(chan, AST_SOFTHANGUP_DEV);
blob = ast_json_pack("{s: s}", "type", "hangup_request"); publish_channel_blob(chan, ast_channel_hangup_request_type(), NULL);
publish_channel_blob(chan, blob);
res = ast_queue_frame(chan, &f); res = ast_queue_frame(chan, &f);
ast_channel_unlock(chan); ast_channel_unlock(chan);
@ -1416,10 +1415,9 @@ int ast_queue_hangup_with_cause(struct ast_channel *chan, int cause)
if (cause < 0) { if (cause < 0) {
f.data.uint32 = ast_channel_hangupcause(chan); f.data.uint32 = ast_channel_hangupcause(chan);
} }
blob = ast_json_pack("{s: s, s: i}", blob = ast_json_pack("{s: i}",
"type", "hangup_request",
"cause", cause); "cause", cause);
publish_channel_blob(chan, blob); publish_channel_blob(chan, ast_channel_hangup_request_type(), blob);
res = ast_queue_frame(chan, &f); res = ast_queue_frame(chan, &f);
ast_channel_unlock(chan); ast_channel_unlock(chan);
@ -2727,11 +2725,10 @@ int ast_softhangup(struct ast_channel *chan, int cause)
ast_channel_lock(chan); ast_channel_lock(chan);
res = ast_softhangup_nolock(chan, cause); res = ast_softhangup_nolock(chan, cause);
blob = ast_json_pack("{s: s, s: i, s: b}", blob = ast_json_pack("{s: i, s: b}",
"type", "hangup_request",
"cause", cause, "cause", cause,
"soft", 1); "soft", 1);
publish_channel_blob(chan, blob); publish_channel_blob(chan, ast_channel_hangup_request_type(), blob);
ast_channel_unlock(chan); ast_channel_unlock(chan);
return res; return res;
@ -3737,15 +3734,14 @@ static void send_dtmf_begin_event(struct ast_channel *chan,
RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref); RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
char digit_str[] = { digit, '\0' }; char digit_str[] = { digit, '\0' };
blob = ast_json_pack("{ s: s, s: s, s: s }", blob = ast_json_pack("{ s: s, s: s }",
"type", "dtmf_begin",
"digit", digit_str, "digit", digit_str,
"direction", dtmf_direction_to_string(direction)); "direction", dtmf_direction_to_string(direction));
if (!blob) { if (!blob) {
return; return;
} }
publish_channel_blob(chan, blob); publish_channel_blob(chan, ast_channel_dtmf_begin_type(), blob);
} }
static void send_dtmf_end_event(struct ast_channel *chan, static void send_dtmf_end_event(struct ast_channel *chan,
@ -3754,8 +3750,7 @@ static void send_dtmf_end_event(struct ast_channel *chan,
RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref); RAII_VAR(struct ast_json *, blob, NULL, ast_json_unref);
char digit_str[] = { digit, '\0' }; char digit_str[] = { digit, '\0' };
blob = ast_json_pack("{ s: s, s: s, s: s, s: i }", blob = ast_json_pack("{ s: s, s: s, s: i }",
"type", "dtmf_end",
"digit", digit_str, "digit", digit_str,
"direction", dtmf_direction_to_string(direction), "direction", dtmf_direction_to_string(direction),
"duration_ms", duration_ms); "duration_ms", duration_ms);
@ -3763,7 +3758,7 @@ static void send_dtmf_end_event(struct ast_channel *chan,
return; return;
} }
publish_channel_blob(chan, blob); publish_channel_blob(chan, ast_channel_dtmf_end_type(), blob);
} }
static void ast_read_generator_actions(struct ast_channel *chan, struct ast_frame *f) static void ast_read_generator_actions(struct ast_channel *chan, struct ast_frame *f)

75
main/manager_channels.c
Unescape View File

@ -544,8 +544,10 @@ static void channel_snapshot_update(void *data, struct stasis_subscription *sub,
} }
} }
static void channel_varset(struct ast_channel_blob *obj) static void channel_varset_cb(void *data, struct stasis_subscription *sub,
struct stasis_topic *topic, struct stasis_message *message)
{ {
struct ast_channel_blob *obj = stasis_message_data(message);
RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free); RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
const char *variable = ast_json_string_get(ast_json_object_get(obj->blob, "variable")); const char *variable = ast_json_string_get(ast_json_object_get(obj->blob, "variable"));
const char *value = ast_json_string_get(ast_json_object_get(obj->blob, "value")); const char *value = ast_json_string_get(ast_json_object_get(obj->blob, "value"));
@ -585,8 +587,10 @@ static void channel_varset(struct ast_channel_blob *obj)
variable, value); variable, value);
} }
static void channel_userevent(struct ast_channel_blob *obj) static void channel_user_event_cb(void *data, struct stasis_subscription *sub,
struct stasis_topic *topic, struct stasis_message *message)
{ {
struct ast_channel_blob *obj = stasis_message_data(message);
RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free); RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
const char *eventname; const char *eventname;
const char *body; const char *body;
@ -620,8 +624,11 @@ static void channel_userevent(struct ast_channel_blob *obj)
ast_str_buffer(channel_event_string), eventname, body); ast_str_buffer(channel_event_string), eventname, body);
} }
static void channel_hangup_request(struct ast_channel_blob *obj) static void channel_hangup_request_cb(void *data,
struct stasis_subscription *sub, struct stasis_topic *topic,
struct stasis_message *message)
{ {
struct ast_channel_blob *obj = stasis_message_data(message);
RAII_VAR(struct ast_str *, extra, NULL, ast_free); RAII_VAR(struct ast_str *, extra, NULL, ast_free);
RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free); RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
struct ast_json *cause; struct ast_json *cause;
@ -657,8 +664,10 @@ static void channel_hangup_request(struct ast_channel_blob *obj)
ast_str_buffer(extra)); ast_str_buffer(extra));
} }
static void channel_dtmf_begin(struct ast_channel_blob *obj) static void channel_dtmf_begin_cb(void *data, struct stasis_subscription *sub,
struct stasis_topic *topic, struct stasis_message *message)
{ {
struct ast_channel_blob *obj = stasis_message_data(message);
RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free); RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
const char *digit = const char *digit =
ast_json_string_get(ast_json_object_get(obj->blob, "digit")); ast_json_string_get(ast_json_object_get(obj->blob, "digit"));
@ -696,8 +705,10 @@ static void channel_dtmf_begin(struct ast_channel_blob *obj)
digit, direction); digit, direction);
} }
static void channel_dtmf_end(struct ast_channel_blob *obj) static void channel_dtmf_end_cb(void *data, struct stasis_subscription *sub,
struct stasis_topic *topic, struct stasis_message *message)
{ {
struct ast_channel_blob *obj = stasis_message_data(message);
RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free); RAII_VAR(struct ast_str *, channel_event_string, NULL, ast_free);
const char *digit = const char *digit =
ast_json_string_get(ast_json_object_get(obj->blob, "digit")); ast_json_string_get(ast_json_object_get(obj->blob, "digit"));
@ -741,34 +752,11 @@ static void channel_dtmf_end(struct ast_channel_blob *obj)
digit, duration_ms, direction); digit, duration_ms, direction);
} }
/*!
* \brief Callback processing messages on the channel topic.
*/
static void channel_blob_cb(void *data, struct stasis_subscription *sub,
struct stasis_topic *topic,
struct stasis_message *message)
{
struct ast_channel_blob *obj = stasis_message_data(message);
if (strcmp("varset", ast_channel_blob_json_type(obj)) == 0) {
channel_varset(obj);
} else if (strcmp("userevent", ast_channel_blob_json_type(obj)) == 0) {
channel_userevent(obj);
} else if (strcmp("hangup_request", ast_channel_blob_json_type(obj)) == 0) {
channel_hangup_request(obj);
} else if (strcmp("dtmf_begin", ast_channel_blob_json_type(obj)) == 0) {
channel_dtmf_begin(obj);
} else if (strcmp("dtmf_end", ast_channel_blob_json_type(obj)) == 0) {
channel_dtmf_end(obj);
}
}
/*! /*!
* \brief Callback processing messages for channel dialing * \brief Callback processing messages for channel dialing
*/ */
static void channel_dial_cb(void *data, struct stasis_subscription *sub, static void channel_dial_cb(void *data, struct stasis_subscription *sub,
struct stasis_topic *topic, struct stasis_topic *topic, struct stasis_message *message)
struct stasis_message *message)
{ {
struct ast_multi_channel_blob *obj = stasis_message_data(message); struct ast_multi_channel_blob *obj = stasis_message_data(message);
const char *dialstatus; const char *dialstatus;
@ -778,11 +766,6 @@ static void channel_dial_cb(void *data, struct stasis_subscription *sub,
RAII_VAR(struct ast_str *, caller_event_string, NULL, ast_free); RAII_VAR(struct ast_str *, caller_event_string, NULL, ast_free);
RAII_VAR(struct ast_str *, peer_event_string, NULL, ast_free); RAII_VAR(struct ast_str *, peer_event_string, NULL, ast_free);
if (strcmp("dial", ast_multi_channel_blob_get_type(obj))) {
ast_assert(0);
return;
}
caller = ast_multi_channel_blob_get_channel(obj, "caller"); caller = ast_multi_channel_blob_get_channel(obj, "caller");
peer = ast_multi_channel_blob_get_channel(obj, "peer"); peer = ast_multi_channel_blob_get_channel(obj, "peer");
@ -852,8 +835,28 @@ int manager_channels_init(void)
NULL); NULL);
ret |= stasis_message_router_add(channel_state_router, ret |= stasis_message_router_add(channel_state_router,
ast_channel_blob_type(), ast_channel_varset_type(),
channel_blob_cb, channel_varset_cb,
NULL);
ret |= stasis_message_router_add(channel_state_router,
ast_channel_user_event_type(),
channel_user_event_cb,
NULL);
ret |= stasis_message_router_add(channel_state_router,
ast_channel_dtmf_begin_type(),
channel_dtmf_begin_cb,
NULL);
ret |= stasis_message_router_add(channel_state_router,
ast_channel_dtmf_end_type(),
channel_dtmf_end_cb,
NULL);
ret |= stasis_message_router_add(channel_state_router,
ast_channel_hangup_request_type(),
channel_hangup_request_cb,
NULL); NULL);
ret |= stasis_message_router_add(channel_state_router, ret |= stasis_message_router_add(channel_state_router,

99
main/stasis_channels.c
Unescape View File

@ -38,14 +38,17 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
#define NUM_MULTI_CHANNEL_BLOB_BUCKETS 7 #define NUM_MULTI_CHANNEL_BLOB_BUCKETS 7
/*! \brief Message type for channel snapshot messages */ /*!
static struct stasis_message_type *channel_snapshot_type; * @{ \brief Define channel message types.
*/
/*! \brief Message type for channel blob messages */ STASIS_MESSAGE_TYPE_DEFN(ast_channel_snapshot_type);
static struct stasis_message_type *channel_blob_type; STASIS_MESSAGE_TYPE_DEFN(ast_channel_dial_type);
STASIS_MESSAGE_TYPE_DEFN(ast_channel_varset_type);
/*! \brief Message type for channel dial messages */ STASIS_MESSAGE_TYPE_DEFN(ast_channel_user_event_type);
static struct stasis_message_type *channel_dial_type; STASIS_MESSAGE_TYPE_DEFN(ast_channel_hangup_request_type);
STASIS_MESSAGE_TYPE_DEFN(ast_channel_dtmf_begin_type);
STASIS_MESSAGE_TYPE_DEFN(ast_channel_dtmf_end_type);
/*! @} */
/*! \brief Topic for all channels */ /*! \brief Topic for all channels */
struct stasis_topic *channel_topic_all; struct stasis_topic *channel_topic_all;
@ -53,21 +56,6 @@ struct stasis_topic *channel_topic_all;
/*! \brief Caching topic for all channels */ /*! \brief Caching topic for all channels */
struct stasis_caching_topic *channel_topic_all_cached; struct stasis_caching_topic *channel_topic_all_cached;
struct stasis_message_type *ast_channel_dial_type(void)
{
return channel_dial_type;
}
struct stasis_message_type *ast_channel_blob_type(void)
{
return channel_blob_type;
}
struct stasis_message_type *ast_channel_snapshot_type(void)
{
return channel_snapshot_type;
}
struct stasis_topic *ast_channel_topic_all(void) struct stasis_topic *ast_channel_topic_all(void)
{ {
return channel_topic_all; return channel_topic_all;
@ -221,18 +209,13 @@ void ast_channel_publish_dial(struct ast_channel *caller, struct ast_channel *pe
} }
struct stasis_message *ast_channel_blob_create(struct ast_channel *chan, struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
struct ast_json *blob) struct stasis_message_type *type, struct ast_json *blob)
{ {
RAII_VAR(struct ast_channel_blob *, obj, NULL, ao2_cleanup); RAII_VAR(struct ast_channel_blob *, obj, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup); RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
struct ast_json *type;
ast_assert(blob != NULL); if (blob == NULL) {
blob = ast_json_null();
type = ast_json_object_get(blob, "type");
if (type == NULL) {
ast_log(LOG_ERROR, "Invalid ast_channel_blob; missing type field\n");
return NULL;
} }
obj = ao2_alloc(sizeof(*obj), channel_blob_dtor); obj = ao2_alloc(sizeof(*obj), channel_blob_dtor);
@ -249,7 +232,7 @@ struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
obj->blob = ast_json_ref(blob); obj->blob = ast_json_ref(blob);
msg = stasis_message_create(ast_channel_blob_type(), obj); msg = stasis_message_create(type, obj);
if (!msg) { if (!msg) {
return NULL; return NULL;
} }
@ -258,15 +241,6 @@ struct stasis_message *ast_channel_blob_create(struct ast_channel *chan,
return msg; return msg;
} }
const char *ast_channel_blob_json_type(struct ast_channel_blob *obj)
{
if (obj == NULL) {
return NULL;
}
return ast_json_string_get(ast_json_object_get(obj->blob, "type"));
}
/*! \brief A channel snapshot wrapper object used in \ref ast_multi_channel_blob objects */ /*! \brief A channel snapshot wrapper object used in \ref ast_multi_channel_blob objects */
struct channel_role_snapshot { struct channel_role_snapshot {
struct ast_channel_snapshot *snapshot; /*!< A channel snapshot */ struct ast_channel_snapshot *snapshot; /*!< A channel snapshot */
@ -319,7 +293,6 @@ struct ast_multi_channel_blob *ast_multi_channel_blob_create(struct ast_json *bl
RAII_VAR(struct ast_multi_channel_blob *, obj, RAII_VAR(struct ast_multi_channel_blob *, obj,
ao2_alloc(sizeof(*obj), multi_channel_blob_dtor), ao2_alloc(sizeof(*obj), multi_channel_blob_dtor),
ao2_cleanup); ao2_cleanup);
struct ast_json *type;
ast_assert(blob != NULL); ast_assert(blob != NULL);
@ -327,12 +300,6 @@ struct ast_multi_channel_blob *ast_multi_channel_blob_create(struct ast_json *bl
return NULL; return NULL;
} }
type = ast_json_object_get(blob, "type");
if (type == NULL) {
ast_log(LOG_ERROR, "Invalid ast_multi_channel_blob; missing type field\n");
return NULL;
}
obj->channel_snapshots = ao2_container_alloc(NUM_MULTI_CHANNEL_BLOB_BUCKETS, obj->channel_snapshots = ao2_container_alloc(NUM_MULTI_CHANNEL_BLOB_BUCKETS,
channel_role_hash_cb, channel_role_single_cmp_cb); channel_role_hash_cb, channel_role_single_cmp_cb);
if (!obj->channel_snapshots) { if (!obj->channel_snapshots) {
@ -423,15 +390,6 @@ struct ast_json *ast_multi_channel_blob_get_json(struct ast_multi_channel_blob *
return obj->blob; return obj->blob;
} }
const char *ast_multi_channel_blob_get_type(struct ast_multi_channel_blob *obj)
{
if (!obj) {
return NULL;
}
return ast_json_string_get(ast_json_object_get(obj->blob, "type"));
}
void ast_channel_publish_varset(struct ast_channel *chan, const char *name, const char *value) void ast_channel_publish_varset(struct ast_channel *chan, const char *name, const char *value)
{ {
RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup); RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
@ -449,7 +407,8 @@ void ast_channel_publish_varset(struct ast_channel *chan, const char *name, cons
return; return;
} }
msg = ast_channel_blob_create(chan, ast_json_ref(blob)); msg = ast_channel_blob_create(chan, ast_channel_varset_type(),
ast_json_ref(blob));
if (!msg) { if (!msg) {
return; return;
@ -491,12 +450,13 @@ struct ast_json *ast_channel_snapshot_to_json(const struct ast_channel_snapshot
void ast_stasis_channels_shutdown(void) void ast_stasis_channels_shutdown(void)
{ {
ao2_cleanup(channel_snapshot_type); STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_snapshot_type);
channel_snapshot_type = NULL; STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_dial_type);
ao2_cleanup(channel_blob_type); STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_varset_type);
channel_blob_type = NULL; STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_user_event_type);
ao2_cleanup(channel_dial_type); STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_hangup_request_type);
channel_dial_type = NULL; STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_dtmf_begin_type);
STASIS_MESSAGE_TYPE_CLEANUP(ast_channel_dtmf_end_type);
ao2_cleanup(channel_topic_all); ao2_cleanup(channel_topic_all);
channel_topic_all = NULL; channel_topic_all = NULL;
channel_topic_all_cached = stasis_caching_unsubscribe(channel_topic_all_cached); channel_topic_all_cached = stasis_caching_unsubscribe(channel_topic_all_cached);
@ -504,9 +464,14 @@ void ast_stasis_channels_shutdown(void)
void ast_stasis_channels_init(void) void ast_stasis_channels_init(void)
{ {
channel_snapshot_type = stasis_message_type_create("ast_channel_snapshot"); STASIS_MESSAGE_TYPE_INIT(ast_channel_snapshot_type);
channel_blob_type = stasis_message_type_create("ast_channel_blob"); STASIS_MESSAGE_TYPE_INIT(ast_channel_dial_type);
channel_dial_type = stasis_message_type_create("ast_channel_dial"); STASIS_MESSAGE_TYPE_INIT(ast_channel_varset_type);
STASIS_MESSAGE_TYPE_INIT(ast_channel_user_event_type);
STASIS_MESSAGE_TYPE_INIT(ast_channel_hangup_request_type);
STASIS_MESSAGE_TYPE_INIT(ast_channel_dtmf_begin_type);
STASIS_MESSAGE_TYPE_INIT(ast_channel_dtmf_end_type);
channel_topic_all = stasis_topic_create("ast_channel_topic_all"); channel_topic_all = stasis_topic_create("ast_channel_topic_all");
channel_topic_all_cached = stasis_caching_topic_create(channel_topic_all, channel_snapshot_get_id); channel_topic_all_cached = stasis_caching_topic_create(channel_topic_all, channel_snapshot_get_id);
} }

23
res/res_stasis.c
Unescape View File

@ -470,19 +470,17 @@ static void dtmf_handler(struct app *app, struct ast_channel_blob *obj)
app_send(app, msg); app_send(app, msg);
} }
static void blob_handler(struct app *app, struct ast_channel_blob *blob)
{
/* To simplify events, we'll only generate on DTMF end */
if (strcmp(ast_channel_blob_json_type(blob), "dtmf_end") == 0) {
dtmf_handler(app, blob);
}
}
static void sub_handler(void *data, struct stasis_subscription *sub, static void sub_handler(void *data, struct stasis_subscription *sub,
struct stasis_topic *topic, struct stasis_topic *topic,
struct stasis_message *message) struct stasis_message *message)
{ {
struct app *app = data; struct app *app = data;
if (stasis_subscription_final_message(sub, message)) {
ao2_cleanup(data);
return;
}
if (ast_channel_snapshot_type() == stasis_message_type(message)) { if (ast_channel_snapshot_type() == stasis_message_type(message)) {
RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref); RAII_VAR(struct ast_json *, msg, NULL, ast_json_unref);
struct ast_channel_snapshot *snapshot = struct ast_channel_snapshot *snapshot =
@ -493,13 +491,12 @@ static void sub_handler(void *data, struct stasis_subscription *sub,
return; return;
} }
app_send(app, msg); app_send(app, msg);
} else if (ast_channel_blob_type() == stasis_message_type(message)) { } else if (ast_channel_dtmf_end_type() == stasis_message_type(message)) {
/* To simplify events, we'll only generate on DTMF end */
struct ast_channel_blob *blob = stasis_message_data(message); struct ast_channel_blob *blob = stasis_message_data(message);
blob_handler(app, blob); dtmf_handler(app, blob);
}
if (stasis_subscription_final_message(sub, message)) {
ao2_cleanup(data);
} }
} }
/*! /*!

64
tests/test_stasis_channels.c
Unescape View File

@ -54,6 +54,7 @@ static void safe_channel_release(struct ast_channel *chan)
AST_TEST_DEFINE(channel_blob_create) AST_TEST_DEFINE(channel_blob_create)
{ {
struct ast_channel_blob *blob; struct ast_channel_blob *blob;
RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup); RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
RAII_VAR(struct ast_channel *, chan, NULL, safe_channel_release); RAII_VAR(struct ast_channel *, chan, NULL, safe_channel_release);
RAII_VAR(struct ast_json *, json, NULL, ast_json_unref); RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
@ -70,36 +71,71 @@ AST_TEST_DEFINE(channel_blob_create)
break; break;
} }
type = stasis_message_type_create("test-type");
chan = ast_channel_alloc(0, AST_STATE_DOWN, "100", "Alice", "100", "100", "default", NULL, 0, "TEST/Alice"); chan = ast_channel_alloc(0, AST_STATE_DOWN, "100", "Alice", "100", "100", "default", NULL, 0, "TEST/Alice");
json = ast_json_pack("{s: s}", json = ast_json_pack("{s: s}",
"type", "test"); "foo", "bar");
bad_json = ast_json_pack("{s: s}",
"bad_key", "test");
/* Off nominal creation */ /* Off nominal creation */
ast_test_validate(test, NULL == ast_channel_blob_create(NULL, bad_json)); ast_test_validate(test, NULL == ast_channel_blob_create(chan, NULL, json));
ast_test_validate(test, NULL == ast_channel_blob_create(chan, bad_json));
/* Test for single channel */ /* Test for single channel */
msg = ast_channel_blob_create(chan, json); msg = ast_channel_blob_create(chan, type, json);
ast_test_validate(test, NULL != msg); ast_test_validate(test, NULL != msg);
blob = stasis_message_data(msg); blob = stasis_message_data(msg);
ast_test_validate(test, NULL != blob); ast_test_validate(test, NULL != blob);
ast_test_validate(test, NULL != blob->snapshot); ast_test_validate(test, NULL != blob->snapshot);
ast_test_validate(test, NULL != blob->blob); ast_test_validate(test, NULL != blob->blob);
ast_test_validate(test, 0 == strcmp(ast_channel_blob_json_type(blob), "test")); ast_test_validate(test, type == stasis_message_type(msg));
ast_test_validate(test, 1 == ao2_ref(msg, 0)); ast_test_validate(test, 1 == ao2_ref(msg, 0));
ao2_cleanup(msg); ao2_cleanup(msg);
/* Test for global channels */ /* Test for global channels */
msg = ast_channel_blob_create(NULL, json); msg = ast_channel_blob_create(NULL, type, json);
ast_test_validate(test, NULL != msg); ast_test_validate(test, NULL != msg);
blob = stasis_message_data(msg); blob = stasis_message_data(msg);
ast_test_validate(test, NULL != blob); ast_test_validate(test, NULL != blob);
ast_test_validate(test, NULL == blob->snapshot); ast_test_validate(test, NULL == blob->snapshot);
ast_test_validate(test, NULL != blob->blob); ast_test_validate(test, NULL != blob->blob);
ast_test_validate(test, 0 == strcmp(ast_channel_blob_json_type(blob), "test")); ast_test_validate(test, type == stasis_message_type(msg));
return AST_TEST_PASS;
}
AST_TEST_DEFINE(null_blob)
{
struct ast_channel_blob *blob;
RAII_VAR(struct stasis_message_type *, type, NULL, ao2_cleanup);
RAII_VAR(struct stasis_message *, msg, NULL, ao2_cleanup);
RAII_VAR(struct ast_channel *, chan, NULL, safe_channel_release);
RAII_VAR(struct ast_json *, json, NULL, ast_json_unref);
RAII_VAR(struct ast_json *, bad_json, NULL, ast_json_unref);
switch (cmd) {
case TEST_INIT:
info->name = __func__;
info->category = test_category;
info->summary = "Test creation of ast_channel_blob objects";
info->description = "Test creation of ast_channel_blob objects";
return AST_TEST_NOT_RUN;
case TEST_EXECUTE:
break;
}
type = stasis_message_type_create("test-type");
chan = ast_channel_alloc(0, AST_STATE_DOWN, "100", "Alice", "100", "100", "default", NULL, 0, "TEST/Alice");
json = ast_json_pack("{s: s}",
"foo", "bar");
/* Test for single channel */
msg = ast_channel_blob_create(chan, type, NULL);
ast_test_validate(test, NULL != msg);
blob = stasis_message_data(msg);
ast_test_validate(test, NULL != blob);
ast_test_validate(test, NULL != blob->snapshot);
ast_test_validate(test, ast_json_null() == blob->blob);
ast_test_validate(test, type == stasis_message_type(msg));
return AST_TEST_PASS; return AST_TEST_PASS;
} }
@ -122,17 +158,11 @@ AST_TEST_DEFINE(multi_channel_blob_create)
} }
json = ast_json_pack("{s: s}", json = ast_json_pack("{s: s}",
"type", "test"); "foo", "bar");
bad_json = ast_json_pack("{s: s}",
"bad_key", "test");
/* Off nominal creation */
ast_test_validate(test, NULL == ast_multi_channel_blob_create(bad_json));
/* Test for single channel */ /* Test for single channel */
blob = ast_multi_channel_blob_create(json); blob = ast_multi_channel_blob_create(json);
ast_test_validate(test, NULL != blob); ast_test_validate(test, NULL != blob);
ast_test_validate(test, 0 == strcmp(ast_multi_channel_blob_get_type(blob), "test"));
ast_test_validate(test, NULL != ast_multi_channel_blob_get_json(blob)); ast_test_validate(test, NULL != ast_multi_channel_blob_get_json(blob));
return AST_TEST_PASS; return AST_TEST_PASS;
@ -266,6 +296,7 @@ AST_TEST_DEFINE(channel_snapshot_json)
static int unload_module(void) static int unload_module(void)
{ {
AST_TEST_UNREGISTER(channel_blob_create); AST_TEST_UNREGISTER(channel_blob_create);
AST_TEST_UNREGISTER(null_blob);
AST_TEST_UNREGISTER(multi_channel_blob_create); AST_TEST_UNREGISTER(multi_channel_blob_create);
AST_TEST_UNREGISTER(multi_channel_blob_snapshots); AST_TEST_UNREGISTER(multi_channel_blob_snapshots);
AST_TEST_UNREGISTER(channel_snapshot_json); AST_TEST_UNREGISTER(channel_snapshot_json);
@ -276,6 +307,7 @@ static int unload_module(void)
static int load_module(void) static int load_module(void)
{ {
AST_TEST_REGISTER(channel_blob_create); AST_TEST_REGISTER(channel_blob_create);
AST_TEST_REGISTER(null_blob);
AST_TEST_REGISTER(multi_channel_blob_create); AST_TEST_REGISTER(multi_channel_blob_create);
AST_TEST_REGISTER(multi_channel_blob_snapshots); AST_TEST_REGISTER(multi_channel_blob_snapshots);
AST_TEST_REGISTER(channel_snapshot_json); AST_TEST_REGISTER(channel_snapshot_json);

Write Preview
Loading…
Cancel
Save