Skip to end of metadata
Go to start of metadata

While the basic API's for the Stasis Message Bus are documented using Doxygen, there's still a lot to be said about how to use those API's within Asterisk.

1. Message Bus

The overall stasis-core API can be best described as a publish/subscribe message bus.

Data that needs to be published on the message bus is encapsulated in a stasis_message, which associates the message data (which is simply an AO2 managed void *) along with a stasis_message_type. (The message also has the timeval when it was created, which is generally useful, since messages are received asynchronously).

Messages are published to a stasis_topic. To receive the messages published to a topic, use stasis_subscribe() to create a subscription.

When topics are subscribed to/unsubscribed to, stasis_subscription_change are sent to all of the topic's subscribers. On unsubscribe, the stasis_subscription_change will be the last message received by the unsubscriber, which can be used to kick off any cleanup that might be necessary. The convenience function stasis_subscription_final_message() can be used to check if a message is the final unsubscribe message.

To forward all of the messages from one topic to another, use stasis_forward_all(). This is useful for creating aggregation topics, like ast_channel_topic_all(), which collect all of the messages published to a set of other topics.

1.1. Routing and Caching

In addition to these fundamental concepts, there are a few extra objects in the Stasis Message Bus arsenal.

For subscriptions that deal with many different types of messages, the stasis_message_router can be used to route messages based on type. This gets rid of a lot of ugly looking if/else chains and replaces them with callback dispatch.

Another common use case within Asterisk is to have a subscription that monitors state changes for snapshots that are published to a topic. To aid in this, we have stasis_caching_topic. This is a topic filter, which presents its own topic with filtered and modified messages from an original topic. In the case of the caching topic, it watches for snapshot messages, compares them with their prior values. For these messages, it publishes a stasis_cache_update message with both the old and new values of the snapshot.

1.2. Design Philosophy

By convention (since there's no effective way to enforce this in C), messages are immutable. Not only the stasis_message type itself, but also the data it contains. This allows messages to be shared freely throughout the system without locking. If you find yourself in the situation where you want to modify the contents of a stasis_message, what you actually want to do is copy the message, and change the copy. While on the surface this seems wasteful and inefficient, the gains in reducing lock contention win the day.

Messages are opaque to the message bus, so you don't have to modify stasis.c or stasis.h to add a new topic or message type.

The stasis_topic object is thread safe. Subscribes, unsubscribes and publishes may happen from any thread. There are some ordering guarantees about message delivery, but not many.

  • A stasis_publish() that begins after a stasis_publish() to the same topic completes will be delivered in order.
    • In general, this means that publications from different threads to the same topic are unordered, but publications from the same thread are.
  • Publications to different topics are not ordered.
  • The final message delivered to a subscription will be the stasis_subscription_final_message().

The callback invocations for a stasis_topic are serialized, although invocations can happen out of any thread from the Stasis thread pool. This is a comfortable middle ground between allocating a thread for every subscription (which would result in too many threads in the system) and invoking callbacks in parallel (which can cause insanity). If your particular handling can benefit from being parallelized, then you can dispatch it to a thread pool in your callback.

2. Topics and messages

While Stasis allows for the proliferation of many, many message types, we feel that creating too many message types would complicate the overall API. We have a small number of first class objects, (channel snapshots, bridge snapshots, etc.). If a message needs a first class object plus a small piece of additional data, blob objects are provided which attach a JSON object to the first class object.

The details of topics and messages are documented on doxygen.asterisk.org. Some of the more important items to be aware of:

2.1. First Class Message Object

2.1.1. ast_channel_snapshot

A snapshot of the primary state of a channel, created by ast_channel_snapshot_create().

2.2. Second Class Message Objects

2.2.1. ast_channel_blob

Often times, you find that you want to attach a piece of data to a snapshot for your message. For these cases, there's ast_channel_blob, which associated a channel snapshot with a JSON object (which is required to have a type field identifying it).

These are used to publish messages such as hangup requests, channel variable set events, etc.

2.3. Channel Topics

2.4. ast_channel_topic(chan)

This is the topic to which channel associated messages are published, including channel snapshots.

2.5. ast_channel_topic_all()

This is an aggregation topic, to which all messages published to individual channel topics are forwarded.

2.6. ast_channel_topic_all_cached()

This is a caching topic wrapping ast_channel_topic_all(), which caches ast_channel_snaphshot messages.

3. Sample Code

3.1. Publishing messages

foo.h
foo.c

3.2. Subscribing (no message router)

bar.c

3.3. Subscribing (with message router)

bar2.c
  • No labels