Struct embassy_sync::pubsub::PubSubChannel

pub struct PubSubChannel<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> { /* private fields */ }

A broadcast channel implementation where multiple publishers can send messages to multiple subscribers

Any published message can be read by all subscribers. A publisher can choose how it sends its message.

• With Pub::publish() the publisher has to wait until there is space in the internal message queue.
• With Pub::publish_immediate() the publisher doesn’t await and instead lets the oldest message in the queue drop if necessary. This will cause any Subscriber that missed the message to receive an error to indicate that it has lagged.

Example

// Create the channel. This can be static as well
let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

// This is a generic subscriber with a direct reference to the channel
let mut sub0 = channel.subscriber().unwrap();
// This is a dynamic subscriber with a dynamic (trait object) reference to the channel
let mut sub1 = channel.dyn_subscriber().unwrap();

let pub0 = channel.publisher().unwrap();

// Publish a message, but wait if the queue is full
pub0.publish(42).await;

// Publish a message, but if the queue is full, just kick out the oldest message.
// This may cause some subscribers to miss a message
pub0.publish_immediate(43);

// Wait for a new message. If the subscriber missed a message, the WaitResult will be a Lag result
assert_eq!(sub0.next_message().await, WaitResult::Message(42));
assert_eq!(sub1.next_message().await, WaitResult::Message(42));

// Wait again, but this time ignore any Lag results
assert_eq!(sub0.next_message_pure().await, 43);
assert_eq!(sub1.next_message_pure().await, 43);

// There's also a polling interface
assert_eq!(sub0.try_next_message(), None);
assert_eq!(sub1.try_next_message(), None);

Implementations

impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> PubSubChannel<M, T, CAP, SUBS, PUBS>

pub const fn new() -> Self

Create a new channel

pub fn subscriber(&self) -> Result<Subscriber<'_, M, T, CAP, SUBS, PUBS>, Error>

Create a new subscriber. It will only receive messages that are published after its creation.

If there are no subscriber slots left, an error will be returned.

pub fn dyn_subscriber(&self) -> Result<DynSubscriber<'_, T>, Error>

Create a new subscriber. It will only receive messages that are published after its creation.

If there are no subscriber slots left, an error will be returned.

pub fn publisher(&self) -> Result<Publisher<'_, M, T, CAP, SUBS, PUBS>, Error>

Create a new publisher

If there are no publisher slots left, an error will be returned.

pub fn dyn_publisher(&self) -> Result<DynPublisher<'_, T>, Error>

Create a new publisher

If there are no publisher slots left, an error will be returned.

pub fn immediate_publisher(
    &self
) -> ImmediatePublisher<'_, M, T, CAP, SUBS, PUBS>

Create a new publisher that can only send immediate messages. This kind of publisher does not take up a publisher slot.

pub fn dyn_immediate_publisher(&self) -> DynImmediatePublisher<'_, T>

Create a new publisher that can only send immediate messages. This kind of publisher does not take up a publisher slot.

Trait Implementations

impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> PubSubBehavior<T> for PubSubChannel<M, T, CAP, SUBS, PUBS>

fn get_message_with_context(
    &self,
    next_message_id: &mut u64,
    cx: Option<&mut Context<'_>>
) -> Poll<WaitResult<T>>

Try to get a message from the queue with the given message id.

fn available(&self, next_message_id: u64) -> u64

Get the amount of messages that are between the given the next_message_id and the most recent message. This is not necessarily the amount of messages a subscriber can still received as it may have lagged.

fn publish_with_context(
    &self,
    message: T,
    cx: Option<&mut Context<'_>>
) -> Result<(), T>

Try to publish a message to the queue.

fn publish_immediate(&self, message: T)

Publish a message immediately

fn space(&self) -> usize

The amount of messages that can still be published without having to wait or without having to lag the subscribers

fn unregister_subscriber(&self, subscriber_next_message_id: u64)

Let the channel know that a subscriber has dropped

fn unregister_publisher(&self)

Let the channel know that a publisher has dropped