Receiving messages

In zbus, structures called observers are used to receive messages on one or more zbus channels. There are multiple types of observers. All modules with a thread use message subscribers, which receive messages asynchronously in the module's own thread.

A message subscriber queues up messages that are received while the module is busy processing another message. The module then processes the messages in the order they were received. An incoming message can never interrupt the processing of another message.

A message subscriber is defined using ZBUS_MSG_SUBSCRIBER_DEFINE, and the subscriber is added to a channel using ZBUS_CHAN_ADD_OBS. For example, in the Network module:

ZBUS_MSG_SUBSCRIBER_DEFINE(network);
ZBUS_CHAN_ADD_OBS(network_chan, network, 0);

The messages are received in the module's thread loop by calling zbus_sub_wait_msg():

err = zbus_sub_wait_msg(&network, &network_state.chan,
            network_state.msg_buf, zbus_wait);

The channel and the message contents are stored in the module's state object in preparation to run the state handler, where the message will be processed.

In modules that subscribe to multiple channels, a channel list macro is used to simplify subscribing to all the channels. The same macro is used to determine the size of a message buffer to handle the largest possible message. For example, in src/main.c:

#define CHANNEL_LIST(X) \
    X(network_chan, struct network_msg) \
    X(cloud_chan, struct cloud_msg) \
    IF_ENABLED(CONFIG_APP_FOTA, (X(fota_chan, struct fota_msg))) \
    IF_ENABLED(CONFIG_APP_LOCATION, (X(location_chan, struct location_msg))) \
    X(main_priv_chan, struct priv_main_msg)

/* Calculate the maximum message size from the list of channels */
#define MAX_MSG_SIZE            MAX_MSG_SIZE_FROM_LIST(CHANNEL_LIST)

/* Add main_subscriber as observer to all the channels in the list. */
#define ADD_OBSERVERS(_chan, _type) ZBUS_CHAN_ADD_OBS(_chan, main_subscriber, 0);

CHANNEL_LIST(ADD_OBSERVERS)