crates_std/vendor/embassy-sync-0.1.0/src/pubsub/mod.rs - third_party/rust_crates - Git at Google

 //! Implementation of [PubSubChannel], a queue where published messages get received by all subscribers.

 #![deny(missing_docs)]

 use core::cell::RefCell;
 use core::fmt::Debug;
 use core::task::{Context, Poll, Waker};

 use heapless::Deque;

 use self::publisher::{ImmediatePub, Pub};
 use self::subscriber::Sub;
 use crate::blocking_mutex::raw::RawMutex;
 use crate::blocking_mutex::Mutex;
 use crate::waitqueue::MultiWakerRegistration;

 pub mod publisher;
 pub mod subscriber;

 pub use publisher::{DynImmediatePublisher, DynPublisher, ImmediatePublisher, Publisher};
 pub use subscriber::{DynSubscriber, Subscriber};

 /// 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
 ///
 /// ```
 /// # use embassy_sync::blocking_mutex::raw::NoopRawMutex;
 /// # use embassy_sync::pubsub::WaitResult;
 /// # use embassy_sync::pubsub::PubSubChannel;
 /// # use futures_executor::block_on;
 /// # let test = async {
 /// // 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);
 /// # };
 /// #
 /// # block_on(test);
 /// ```
 ///
 pub struct PubSubChannel<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> {
     inner: Mutex<M, RefCell<PubSubState<T, CAP, SUBS, PUBS>>>,
 }

 impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize>
     PubSubChannel<M, T, CAP, SUBS, PUBS>
 {
     /// Create a new channel
     pub const fn new() -> Self {
         Self {
             inner: Mutex::const_new(M::INIT, RefCell::new(PubSubState::new())),
         }
     }

     /// 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 subscriber(&self) -> Result<Subscriber<M, T, CAP, SUBS, PUBS>, Error> {
         self.inner.lock(|inner| {
             let mut s = inner.borrow_mut();

             if s.subscriber_count >= SUBS {
                 Err(Error::MaximumSubscribersReached)
             } else {
                 s.subscriber_count += 1;
                 Ok(Subscriber(Sub::new(s.next_message_id, self)))
             }
         })
     }

     /// 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> {
         self.inner.lock(|inner| {
             let mut s = inner.borrow_mut();

             if s.subscriber_count >= SUBS {
                 Err(Error::MaximumSubscribersReached)
             } else {
                 s.subscriber_count += 1;
                 Ok(DynSubscriber(Sub::new(s.next_message_id, self)))
             }
         })
     }

     /// Create a new publisher
     ///
     /// If there are no publisher slots left, an error will be returned.
     pub fn publisher(&self) -> Result<Publisher<M, T, CAP, SUBS, PUBS>, Error> {
         self.inner.lock(|inner| {
             let mut s = inner.borrow_mut();

             if s.publisher_count >= PUBS {
                 Err(Error::MaximumPublishersReached)
             } else {
                 s.publisher_count += 1;
                 Ok(Publisher(Pub::new(self)))
             }
         })
     }

     /// 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> {
         self.inner.lock(|inner| {
             let mut s = inner.borrow_mut();

             if s.publisher_count >= PUBS {
                 Err(Error::MaximumPublishersReached)
             } else {
                 s.publisher_count += 1;
                 Ok(DynPublisher(Pub::new(self)))
             }
         })
     }

     /// Create a new publisher that can only send immediate messages.
     /// This kind of publisher does not take up a publisher slot.
     pub fn immediate_publisher(&self) -> ImmediatePublisher<M, T, CAP, SUBS, PUBS> {
         ImmediatePublisher(ImmediatePub::new(self))
     }

     /// 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> {
         DynImmediatePublisher(ImmediatePub::new(self))
     }
 }

 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>> {
         self.inner.lock(|s| {
             let mut s = s.borrow_mut();

             // Check if we can read a message
             match s.get_message(*next_message_id) {
                 // Yes, so we are done polling
                 Some(WaitResult::Message(message)) => {
                     *next_message_id += 1;
                     Poll::Ready(WaitResult::Message(message))
                 }
                 // No, so we need to reregister our waker and sleep again
                 None => {
                     if let Some(cx) = cx {
                         s.register_subscriber_waker(cx.waker());
                     }
                     Poll::Pending
                 }
                 // We missed a couple of messages. We must do our internal bookkeeping and return that we lagged
                 Some(WaitResult::Lagged(amount)) => {
                     *next_message_id += amount;
                     Poll::Ready(WaitResult::Lagged(amount))
                 }
             }
         })
     }

     fn available(&self, next_message_id: u64) -> u64 {
         self.inner.lock(|s| s.borrow().next_message_id - next_message_id)
     }

     fn publish_with_context(&self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), T> {
         self.inner.lock(|s| {
             let mut s = s.borrow_mut();
             // Try to publish the message
             match s.try_publish(message) {
                 // We did it, we are ready
                 Ok(()) => Ok(()),
                 // The queue is full, so we need to reregister our waker and go to sleep
                 Err(message) => {
                     if let Some(cx) = cx {
                         s.register_publisher_waker(cx.waker());
                     }
                     Err(message)
                 }
             }
         })
     }

     fn publish_immediate(&self, message: T) {
         self.inner.lock(|s| {
             let mut s = s.borrow_mut();
             s.publish_immediate(message)
         })
     }

     fn space(&self) -> usize {
         self.inner.lock(|s| {
             let s = s.borrow();
             s.queue.capacity() - s.queue.len()
         })
     }

     fn unregister_subscriber(&self, subscriber_next_message_id: u64) {
         self.inner.lock(|s| {
             let mut s = s.borrow_mut();
             s.unregister_subscriber(subscriber_next_message_id)
         })
     }

     fn unregister_publisher(&self) {
         self.inner.lock(|s| {
             let mut s = s.borrow_mut();
             s.unregister_publisher()
         })
     }
 }

 /// Internal state for the PubSub channel
 struct PubSubState<T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> {
     /// The queue contains the last messages that have been published and a countdown of how many subscribers are yet to read it
     queue: Deque<(T, usize), CAP>,
     /// Every message has an id.
     /// Don't worry, we won't run out.
     /// If a million messages were published every second, then the ID's would run out in about 584942 years.
     next_message_id: u64,
     /// Collection of wakers for Subscribers that are waiting.
     subscriber_wakers: MultiWakerRegistration<SUBS>,
     /// Collection of wakers for Publishers that are waiting.
     publisher_wakers: MultiWakerRegistration<PUBS>,
     /// The amount of subscribers that are active
     subscriber_count: usize,
     /// The amount of publishers that are active
     publisher_count: usize,
 }

 impl<T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> PubSubState<T, CAP, SUBS, PUBS> {
     /// Create a new internal channel state
     const fn new() -> Self {
         Self {
             queue: Deque::new(),
             next_message_id: 0,
             subscriber_wakers: MultiWakerRegistration::new(),
             publisher_wakers: MultiWakerRegistration::new(),
             subscriber_count: 0,
             publisher_count: 0,
         }
     }

     fn try_publish(&mut self, message: T) -> Result<(), T> {
         if self.subscriber_count == 0 {
             // We don't need to publish anything because there is no one to receive it
             return Ok(());
         }

         if self.queue.is_full() {
             return Err(message);
         }
         // We just did a check for this
         self.queue.push_back((message, self.subscriber_count)).ok().unwrap();

         self.next_message_id += 1;

         // Wake all of the subscribers
         self.subscriber_wakers.wake();

         Ok(())
     }

     fn publish_immediate(&mut self, message: T) {
         // Make space in the queue if required
         if self.queue.is_full() {
             self.queue.pop_front();
         }

         // This will succeed because we made sure there is space
         self.try_publish(message).ok().unwrap();
     }

     fn get_message(&mut self, message_id: u64) -> Option<WaitResult<T>> {
         let start_id = self.next_message_id - self.queue.len() as u64;

         if message_id < start_id {
             return Some(WaitResult::Lagged(start_id - message_id));
         }

         let current_message_index = (message_id - start_id) as usize;

         if current_message_index >= self.queue.len() {
             return None;
         }

         // We've checked that the index is valid
         let queue_item = self.queue.iter_mut().nth(current_message_index).unwrap();

         // We're reading this item, so decrement the counter
         queue_item.1 -= 1;
         let message = queue_item.0.clone();

         if current_message_index == 0 && queue_item.1 == 0 {
             self.queue.pop_front();
             self.publisher_wakers.wake();
         }

         Some(WaitResult::Message(message))
     }

     fn register_subscriber_waker(&mut self, waker: &Waker) {
         match self.subscriber_wakers.register(waker) {
             Ok(()) => {}
             Err(_) => {
                 // All waker slots were full. This can only happen when there was a subscriber that now has dropped.
                 // We need to throw it away. It's a bit inefficient, but we can wake everything.
                 // Any future that is still active will simply reregister.
                 // This won't happen a lot, so it's ok.
                 self.subscriber_wakers.wake();
                 self.subscriber_wakers.register(waker).unwrap();
             }
         }
     }

     fn register_publisher_waker(&mut self, waker: &Waker) {
         match self.publisher_wakers.register(waker) {
             Ok(()) => {}
             Err(_) => {
                 // All waker slots were full. This can only happen when there was a publisher that now has dropped.
                 // We need to throw it away. It's a bit inefficient, but we can wake everything.
                 // Any future that is still active will simply reregister.
                 // This won't happen a lot, so it's ok.
                 self.publisher_wakers.wake();
                 self.publisher_wakers.register(waker).unwrap();
             }
         }
     }

     fn unregister_subscriber(&mut self, subscriber_next_message_id: u64) {
         self.subscriber_count -= 1;

         // All messages that haven't been read yet by this subscriber must have their counter decremented
         let start_id = self.next_message_id - self.queue.len() as u64;
         if subscriber_next_message_id >= start_id {
             let current_message_index = (subscriber_next_message_id - start_id) as usize;
             self.queue
                 .iter_mut()
                 .skip(current_message_index)
                 .for_each(|(_, counter)| *counter -= 1);
         }
     }

     fn unregister_publisher(&mut self) {
         self.publisher_count -= 1;
     }
 }

 /// Error type for the [PubSubChannel]
 #[derive(Debug, PartialEq, Eq, Clone)]
 #[cfg_attr(feature = "defmt", derive(defmt::Format))]
 pub enum Error {
     /// All subscriber slots are used. To add another subscriber, first another subscriber must be dropped or
     /// the capacity of the channels must be increased.
     MaximumSubscribersReached,
     /// All publisher slots are used. To add another publisher, first another publisher must be dropped or
     /// the capacity of the channels must be increased.
     MaximumPublishersReached,
 }

 /// 'Middle level' behaviour of the pubsub channel.
 /// This trait is used so that Sub and Pub can be generic over the channel.
 pub trait PubSubBehavior<T> {
     /// Try to get a message from the queue with the given message id.
     ///
     /// If the message is not yet present and a context is given, then its waker is registered in the subsriber wakers.
     fn get_message_with_context(&self, next_message_id: &mut u64, cx: Option<&mut Context<'_>>) -> Poll<WaitResult<T>>;

     /// 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 available(&self, next_message_id: u64) -> u64;

     /// Try to publish a message to the queue.
     ///
     /// If the queue is full and a context is given, then its waker is registered in the publisher wakers.
     fn publish_with_context(&self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), T>;

     /// Publish a message immediately
     fn publish_immediate(&self, message: T);

     /// The amount of messages that can still be published without having to wait or without having to lag the subscribers
     fn space(&self) -> usize;

     /// Let the channel know that a subscriber has dropped
     fn unregister_subscriber(&self, subscriber_next_message_id: u64);

     /// Let the channel know that a publisher has dropped
     fn unregister_publisher(&self);
 }

 /// The result of the subscriber wait procedure
 #[derive(Debug, Clone, PartialEq, Eq)]
 #[cfg_attr(feature = "defmt", derive(defmt::Format))]
 pub enum WaitResult<T> {
     /// The subscriber did not receive all messages and lagged by the given amount of messages.
     /// (This is the amount of messages that were missed)
     Lagged(u64),
     /// A message was received
     Message(T),
 }

 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::blocking_mutex::raw::NoopRawMutex;

     #[futures_test::test]
     async fn dyn_pub_sub_works() {
         let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

         let mut sub0 = channel.dyn_subscriber().unwrap();
         let mut sub1 = channel.dyn_subscriber().unwrap();
         let pub0 = channel.dyn_publisher().unwrap();

         pub0.publish(42).await;

         assert_eq!(sub0.next_message().await, WaitResult::Message(42));
         assert_eq!(sub1.next_message().await, WaitResult::Message(42));

         assert_eq!(sub0.try_next_message(), None);
         assert_eq!(sub1.try_next_message(), None);
     }

     #[futures_test::test]
     async fn all_subscribers_receive() {
         let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

         let mut sub0 = channel.subscriber().unwrap();
         let mut sub1 = channel.subscriber().unwrap();
         let pub0 = channel.publisher().unwrap();

         pub0.publish(42).await;

         assert_eq!(sub0.next_message().await, WaitResult::Message(42));
         assert_eq!(sub1.next_message().await, WaitResult::Message(42));

         assert_eq!(sub0.try_next_message(), None);
         assert_eq!(sub1.try_next_message(), None);
     }

     #[futures_test::test]
     async fn lag_when_queue_full_on_immediate_publish() {
         let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

         let mut sub0 = channel.subscriber().unwrap();
         let pub0 = channel.publisher().unwrap();

         pub0.publish_immediate(42);
         pub0.publish_immediate(43);
         pub0.publish_immediate(44);
         pub0.publish_immediate(45);
         pub0.publish_immediate(46);
         pub0.publish_immediate(47);

         assert_eq!(sub0.try_next_message(), Some(WaitResult::Lagged(2)));
         assert_eq!(sub0.next_message().await, WaitResult::Message(44));
         assert_eq!(sub0.next_message().await, WaitResult::Message(45));
         assert_eq!(sub0.next_message().await, WaitResult::Message(46));
         assert_eq!(sub0.next_message().await, WaitResult::Message(47));
         assert_eq!(sub0.try_next_message(), None);
     }

     #[test]
     fn limited_subs_and_pubs() {
         let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

         let sub0 = channel.subscriber();
         let sub1 = channel.subscriber();
         let sub2 = channel.subscriber();
         let sub3 = channel.subscriber();
         let sub4 = channel.subscriber();

         assert!(sub0.is_ok());
         assert!(sub1.is_ok());
         assert!(sub2.is_ok());
         assert!(sub3.is_ok());
         assert_eq!(sub4.err().unwrap(), Error::MaximumSubscribersReached);

         drop(sub0);

         let sub5 = channel.subscriber();
         assert!(sub5.is_ok());

         // publishers

         let pub0 = channel.publisher();
         let pub1 = channel.publisher();
         let pub2 = channel.publisher();
         let pub3 = channel.publisher();
         let pub4 = channel.publisher();

         assert!(pub0.is_ok());
         assert!(pub1.is_ok());
         assert!(pub2.is_ok());
         assert!(pub3.is_ok());
         assert_eq!(pub4.err().unwrap(), Error::MaximumPublishersReached);

         drop(pub0);

         let pub5 = channel.publisher();
         assert!(pub5.is_ok());
     }

     #[test]
     fn publisher_wait_on_full_queue() {
         let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

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

         // There are no subscribers, so the queue will never be full
         assert_eq!(pub0.try_publish(0), Ok(()));
         assert_eq!(pub0.try_publish(0), Ok(()));
         assert_eq!(pub0.try_publish(0), Ok(()));
         assert_eq!(pub0.try_publish(0), Ok(()));
         assert_eq!(pub0.try_publish(0), Ok(()));

         let sub0 = channel.subscriber().unwrap();

         assert_eq!(pub0.try_publish(0), Ok(()));
         assert_eq!(pub0.try_publish(0), Ok(()));
         assert_eq!(pub0.try_publish(0), Ok(()));
         assert_eq!(pub0.try_publish(0), Ok(()));
         assert_eq!(pub0.try_publish(0), Err(0));

         drop(sub0);
     }

     #[futures_test::test]
     async fn correct_available() {
         let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

         let sub0 = channel.subscriber().unwrap();
         let mut sub1 = channel.subscriber().unwrap();
         let pub0 = channel.publisher().unwrap();

         assert_eq!(sub0.available(), 0);
         assert_eq!(sub1.available(), 0);

         pub0.publish(42).await;

         assert_eq!(sub0.available(), 1);
         assert_eq!(sub1.available(), 1);

         sub1.next_message().await;

         assert_eq!(sub1.available(), 0);

         pub0.publish(42).await;

         assert_eq!(sub0.available(), 2);
         assert_eq!(sub1.available(), 1);
     }

     #[futures_test::test]
     async fn correct_space() {
         let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

         let mut sub0 = channel.subscriber().unwrap();
         let mut sub1 = channel.subscriber().unwrap();
         let pub0 = channel.publisher().unwrap();

         assert_eq!(pub0.space(), 4);

         pub0.publish(42).await;

         assert_eq!(pub0.space(), 3);

         pub0.publish(42).await;

         assert_eq!(pub0.space(), 2);

         sub0.next_message().await;
         sub0.next_message().await;

         assert_eq!(pub0.space(), 2);

         sub1.next_message().await;
         assert_eq!(pub0.space(), 3);
         sub1.next_message().await;
         assert_eq!(pub0.space(), 4);
     }
 }
	//! Implementation of [PubSubChannel], a queue where published messages get received by all subscribers.

	#![deny(missing_docs)]

	use core::cell::RefCell;
	use core::fmt::Debug;
	use core::task::{Context, Poll, Waker};

	use heapless::Deque;

	use self::publisher::{ImmediatePub, Pub};
	use self::subscriber::Sub;
	use crate::blocking_mutex::raw::RawMutex;
	use crate::blocking_mutex::Mutex;
	use crate::waitqueue::MultiWakerRegistration;

	pub mod publisher;
	pub mod subscriber;

	pub use publisher::{DynImmediatePublisher, DynPublisher, ImmediatePublisher, Publisher};
	pub use subscriber::{DynSubscriber, Subscriber};

	/// 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
	///
	/// ```
	/// # use embassy_sync::blocking_mutex::raw::NoopRawMutex;
	/// # use embassy_sync::pubsub::WaitResult;
	/// # use embassy_sync::pubsub::PubSubChannel;
	/// # use futures_executor::block_on;
	/// # let test = async {
	/// // 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);
	/// # };
	/// #
	/// # block_on(test);
	/// ```
	///
	pub struct PubSubChannel<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> {
	inner: Mutex<M, RefCell<PubSubState<T, CAP, SUBS, PUBS>>>,
	}

	impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize>
	PubSubChannel<M, T, CAP, SUBS, PUBS>
	{
	/// Create a new channel
	pub const fn new() -> Self {
	Self {
	inner: Mutex::const_new(M::INIT, RefCell::new(PubSubState::new())),
	}
	}

	/// 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 subscriber(&self) -> Result<Subscriber<M, T, CAP, SUBS, PUBS>, Error> {
	self.inner.lock(\|inner\| {
	let mut s = inner.borrow_mut();

	if s.subscriber_count >= SUBS {
	Err(Error::MaximumSubscribersReached)
	} else {
	s.subscriber_count += 1;
	Ok(Subscriber(Sub::new(s.next_message_id, self)))
	}
	})
	}

	/// 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> {
	self.inner.lock(\|inner\| {
	let mut s = inner.borrow_mut();

	if s.subscriber_count >= SUBS {
	Err(Error::MaximumSubscribersReached)
	} else {
	s.subscriber_count += 1;
	Ok(DynSubscriber(Sub::new(s.next_message_id, self)))
	}
	})
	}

	/// Create a new publisher
	///
	/// If there are no publisher slots left, an error will be returned.
	pub fn publisher(&self) -> Result<Publisher<M, T, CAP, SUBS, PUBS>, Error> {
	self.inner.lock(\|inner\| {
	let mut s = inner.borrow_mut();

	if s.publisher_count >= PUBS {
	Err(Error::MaximumPublishersReached)
	} else {
	s.publisher_count += 1;
	Ok(Publisher(Pub::new(self)))
	}
	})
	}

	/// 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> {
	self.inner.lock(\|inner\| {
	let mut s = inner.borrow_mut();

	if s.publisher_count >= PUBS {
	Err(Error::MaximumPublishersReached)
	} else {
	s.publisher_count += 1;
	Ok(DynPublisher(Pub::new(self)))
	}
	})
	}

	/// Create a new publisher that can only send immediate messages.
	/// This kind of publisher does not take up a publisher slot.
	pub fn immediate_publisher(&self) -> ImmediatePublisher<M, T, CAP, SUBS, PUBS> {
	ImmediatePublisher(ImmediatePub::new(self))
	}

	/// 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> {
	DynImmediatePublisher(ImmediatePub::new(self))
	}
	}

	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>> {
	self.inner.lock(\|s\| {
	let mut s = s.borrow_mut();

	// Check if we can read a message
	match s.get_message(*next_message_id) {
	// Yes, so we are done polling
	Some(WaitResult::Message(message)) => {
	*next_message_id += 1;
	Poll::Ready(WaitResult::Message(message))
	}
	// No, so we need to reregister our waker and sleep again
	None => {
	if let Some(cx) = cx {
	s.register_subscriber_waker(cx.waker());
	}
	Poll::Pending
	}
	// We missed a couple of messages. We must do our internal bookkeeping and return that we lagged
	Some(WaitResult::Lagged(amount)) => {
	*next_message_id += amount;
	Poll::Ready(WaitResult::Lagged(amount))
	}
	}
	})
	}

	fn available(&self, next_message_id: u64) -> u64 {
	self.inner.lock(\|s\| s.borrow().next_message_id - next_message_id)
	}

	fn publish_with_context(&self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), T> {
	self.inner.lock(\|s\| {
	let mut s = s.borrow_mut();
	// Try to publish the message
	match s.try_publish(message) {
	// We did it, we are ready
	Ok(()) => Ok(()),
	// The queue is full, so we need to reregister our waker and go to sleep
	Err(message) => {
	if let Some(cx) = cx {
	s.register_publisher_waker(cx.waker());
	}
	Err(message)
	}
	}
	})
	}

	fn publish_immediate(&self, message: T) {
	self.inner.lock(\|s\| {
	let mut s = s.borrow_mut();
	s.publish_immediate(message)
	})
	}

	fn space(&self) -> usize {
	self.inner.lock(\|s\| {
	let s = s.borrow();
	s.queue.capacity() - s.queue.len()
	})
	}

	fn unregister_subscriber(&self, subscriber_next_message_id: u64) {
	self.inner.lock(\|s\| {
	let mut s = s.borrow_mut();
	s.unregister_subscriber(subscriber_next_message_id)
	})
	}

	fn unregister_publisher(&self) {
	self.inner.lock(\|s\| {
	let mut s = s.borrow_mut();
	s.unregister_publisher()
	})
	}
	}

	/// Internal state for the PubSub channel
	struct PubSubState<T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> {
	/// The queue contains the last messages that have been published and a countdown of how many subscribers are yet to read it
	queue: Deque<(T, usize), CAP>,
	/// Every message has an id.
	/// Don't worry, we won't run out.
	/// If a million messages were published every second, then the ID's would run out in about 584942 years.
	next_message_id: u64,
	/// Collection of wakers for Subscribers that are waiting.
	subscriber_wakers: MultiWakerRegistration<SUBS>,
	/// Collection of wakers for Publishers that are waiting.
	publisher_wakers: MultiWakerRegistration<PUBS>,
	/// The amount of subscribers that are active
	subscriber_count: usize,
	/// The amount of publishers that are active
	publisher_count: usize,
	}

	impl<T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usize> PubSubState<T, CAP, SUBS, PUBS> {
	/// Create a new internal channel state
	const fn new() -> Self {
	Self {
	queue: Deque::new(),
	next_message_id: 0,
	subscriber_wakers: MultiWakerRegistration::new(),
	publisher_wakers: MultiWakerRegistration::new(),
	subscriber_count: 0,
	publisher_count: 0,
	}
	}

	fn try_publish(&mut self, message: T) -> Result<(), T> {
	if self.subscriber_count == 0 {
	// We don't need to publish anything because there is no one to receive it
	return Ok(());
	}

	if self.queue.is_full() {
	return Err(message);
	}
	// We just did a check for this
	self.queue.push_back((message, self.subscriber_count)).ok().unwrap();

	self.next_message_id += 1;

	// Wake all of the subscribers
	self.subscriber_wakers.wake();

	Ok(())
	}

	fn publish_immediate(&mut self, message: T) {
	// Make space in the queue if required
	if self.queue.is_full() {
	self.queue.pop_front();
	}

	// This will succeed because we made sure there is space
	self.try_publish(message).ok().unwrap();
	}

	fn get_message(&mut self, message_id: u64) -> Option<WaitResult<T>> {
	let start_id = self.next_message_id - self.queue.len() as u64;

	if message_id < start_id {
	return Some(WaitResult::Lagged(start_id - message_id));
	}

	let current_message_index = (message_id - start_id) as usize;

	if current_message_index >= self.queue.len() {
	return None;
	}

	// We've checked that the index is valid
	let queue_item = self.queue.iter_mut().nth(current_message_index).unwrap();

	// We're reading this item, so decrement the counter
	queue_item.1 -= 1;
	let message = queue_item.0.clone();

	if current_message_index == 0 && queue_item.1 == 0 {
	self.queue.pop_front();
	self.publisher_wakers.wake();
	}

	Some(WaitResult::Message(message))
	}

	fn register_subscriber_waker(&mut self, waker: &Waker) {
	match self.subscriber_wakers.register(waker) {
	Ok(()) => {}
	Err(_) => {
	// All waker slots were full. This can only happen when there was a subscriber that now has dropped.
	// We need to throw it away. It's a bit inefficient, but we can wake everything.
	// Any future that is still active will simply reregister.
	// This won't happen a lot, so it's ok.
	self.subscriber_wakers.wake();
	self.subscriber_wakers.register(waker).unwrap();
	}
	}
	}

	fn register_publisher_waker(&mut self, waker: &Waker) {
	match self.publisher_wakers.register(waker) {
	Ok(()) => {}
	Err(_) => {
	// All waker slots were full. This can only happen when there was a publisher that now has dropped.
	// We need to throw it away. It's a bit inefficient, but we can wake everything.
	// Any future that is still active will simply reregister.
	// This won't happen a lot, so it's ok.
	self.publisher_wakers.wake();
	self.publisher_wakers.register(waker).unwrap();
	}
	}
	}

	fn unregister_subscriber(&mut self, subscriber_next_message_id: u64) {
	self.subscriber_count -= 1;

	// All messages that haven't been read yet by this subscriber must have their counter decremented
	let start_id = self.next_message_id - self.queue.len() as u64;
	if subscriber_next_message_id >= start_id {
	let current_message_index = (subscriber_next_message_id - start_id) as usize;
	self.queue
	.iter_mut()
	.skip(current_message_index)
	.for_each(\|(_, counter)\| *counter -= 1);
	}
	}

	fn unregister_publisher(&mut self) {
	self.publisher_count -= 1;
	}
	}

	/// Error type for the [PubSubChannel]
	#[derive(Debug, PartialEq, Eq, Clone)]
	#[cfg_attr(feature = "defmt", derive(defmt::Format))]
	pub enum Error {
	/// All subscriber slots are used. To add another subscriber, first another subscriber must be dropped or
	/// the capacity of the channels must be increased.
	MaximumSubscribersReached,
	/// All publisher slots are used. To add another publisher, first another publisher must be dropped or
	/// the capacity of the channels must be increased.
	MaximumPublishersReached,
	}

	/// 'Middle level' behaviour of the pubsub channel.
	/// This trait is used so that Sub and Pub can be generic over the channel.
	pub trait PubSubBehavior<T> {
	/// Try to get a message from the queue with the given message id.
	///
	/// If the message is not yet present and a context is given, then its waker is registered in the subsriber wakers.
	fn get_message_with_context(&self, next_message_id: &mut u64, cx: Option<&mut Context<'_>>) -> Poll<WaitResult<T>>;

	/// 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 available(&self, next_message_id: u64) -> u64;

	/// Try to publish a message to the queue.
	///
	/// If the queue is full and a context is given, then its waker is registered in the publisher wakers.
	fn publish_with_context(&self, message: T, cx: Option<&mut Context<'_>>) -> Result<(), T>;

	/// Publish a message immediately
	fn publish_immediate(&self, message: T);

	/// The amount of messages that can still be published without having to wait or without having to lag the subscribers
	fn space(&self) -> usize;

	/// Let the channel know that a subscriber has dropped
	fn unregister_subscriber(&self, subscriber_next_message_id: u64);

	/// Let the channel know that a publisher has dropped
	fn unregister_publisher(&self);
	}

	/// The result of the subscriber wait procedure
	#[derive(Debug, Clone, PartialEq, Eq)]
	#[cfg_attr(feature = "defmt", derive(defmt::Format))]
	pub enum WaitResult<T> {
	/// The subscriber did not receive all messages and lagged by the given amount of messages.
	/// (This is the amount of messages that were missed)
	Lagged(u64),
	/// A message was received
	Message(T),
	}

	#[cfg(test)]
	mod tests {
	use super::*;
	use crate::blocking_mutex::raw::NoopRawMutex;

	#[futures_test::test]
	async fn dyn_pub_sub_works() {
	let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

	let mut sub0 = channel.dyn_subscriber().unwrap();
	let mut sub1 = channel.dyn_subscriber().unwrap();
	let pub0 = channel.dyn_publisher().unwrap();

	pub0.publish(42).await;

	assert_eq!(sub0.next_message().await, WaitResult::Message(42));
	assert_eq!(sub1.next_message().await, WaitResult::Message(42));

	assert_eq!(sub0.try_next_message(), None);
	assert_eq!(sub1.try_next_message(), None);
	}

	#[futures_test::test]
	async fn all_subscribers_receive() {
	let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

	let mut sub0 = channel.subscriber().unwrap();
	let mut sub1 = channel.subscriber().unwrap();
	let pub0 = channel.publisher().unwrap();

	pub0.publish(42).await;

	assert_eq!(sub0.next_message().await, WaitResult::Message(42));
	assert_eq!(sub1.next_message().await, WaitResult::Message(42));

	assert_eq!(sub0.try_next_message(), None);
	assert_eq!(sub1.try_next_message(), None);
	}

	#[futures_test::test]
	async fn lag_when_queue_full_on_immediate_publish() {
	let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

	let mut sub0 = channel.subscriber().unwrap();
	let pub0 = channel.publisher().unwrap();

	pub0.publish_immediate(42);
	pub0.publish_immediate(43);
	pub0.publish_immediate(44);
	pub0.publish_immediate(45);
	pub0.publish_immediate(46);
	pub0.publish_immediate(47);

	assert_eq!(sub0.try_next_message(), Some(WaitResult::Lagged(2)));
	assert_eq!(sub0.next_message().await, WaitResult::Message(44));
	assert_eq!(sub0.next_message().await, WaitResult::Message(45));
	assert_eq!(sub0.next_message().await, WaitResult::Message(46));
	assert_eq!(sub0.next_message().await, WaitResult::Message(47));
	assert_eq!(sub0.try_next_message(), None);
	}

	#[test]
	fn limited_subs_and_pubs() {
	let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

	let sub0 = channel.subscriber();
	let sub1 = channel.subscriber();
	let sub2 = channel.subscriber();
	let sub3 = channel.subscriber();
	let sub4 = channel.subscriber();

	assert!(sub0.is_ok());
	assert!(sub1.is_ok());
	assert!(sub2.is_ok());
	assert!(sub3.is_ok());
	assert_eq!(sub4.err().unwrap(), Error::MaximumSubscribersReached);

	drop(sub0);

	let sub5 = channel.subscriber();
	assert!(sub5.is_ok());

	// publishers

	let pub0 = channel.publisher();
	let pub1 = channel.publisher();
	let pub2 = channel.publisher();
	let pub3 = channel.publisher();
	let pub4 = channel.publisher();

	assert!(pub0.is_ok());
	assert!(pub1.is_ok());
	assert!(pub2.is_ok());
	assert!(pub3.is_ok());
	assert_eq!(pub4.err().unwrap(), Error::MaximumPublishersReached);

	drop(pub0);

	let pub5 = channel.publisher();
	assert!(pub5.is_ok());
	}

	#[test]
	fn publisher_wait_on_full_queue() {
	let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

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

	// There are no subscribers, so the queue will never be full
	assert_eq!(pub0.try_publish(0), Ok(()));
	assert_eq!(pub0.try_publish(0), Ok(()));
	assert_eq!(pub0.try_publish(0), Ok(()));
	assert_eq!(pub0.try_publish(0), Ok(()));
	assert_eq!(pub0.try_publish(0), Ok(()));

	let sub0 = channel.subscriber().unwrap();

	assert_eq!(pub0.try_publish(0), Ok(()));
	assert_eq!(pub0.try_publish(0), Ok(()));
	assert_eq!(pub0.try_publish(0), Ok(()));
	assert_eq!(pub0.try_publish(0), Ok(()));
	assert_eq!(pub0.try_publish(0), Err(0));

	drop(sub0);
	}

	#[futures_test::test]
	async fn correct_available() {
	let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

	let sub0 = channel.subscriber().unwrap();
	let mut sub1 = channel.subscriber().unwrap();
	let pub0 = channel.publisher().unwrap();

	assert_eq!(sub0.available(), 0);
	assert_eq!(sub1.available(), 0);

	pub0.publish(42).await;

	assert_eq!(sub0.available(), 1);
	assert_eq!(sub1.available(), 1);

	sub1.next_message().await;

	assert_eq!(sub1.available(), 0);

	pub0.publish(42).await;

	assert_eq!(sub0.available(), 2);
	assert_eq!(sub1.available(), 1);
	}

	#[futures_test::test]
	async fn correct_space() {
	let channel = PubSubChannel::<NoopRawMutex, u32, 4, 4, 4>::new();

	let mut sub0 = channel.subscriber().unwrap();
	let mut sub1 = channel.subscriber().unwrap();
	let pub0 = channel.publisher().unwrap();

	assert_eq!(pub0.space(), 4);

	pub0.publish(42).await;

	assert_eq!(pub0.space(), 3);

	pub0.publish(42).await;

	assert_eq!(pub0.space(), 2);

	sub0.next_message().await;
	sub0.next_message().await;

	assert_eq!(pub0.space(), 2);

	sub1.next_message().await;
	assert_eq!(pub0.space(), 3);
	sub1.next_message().await;
	assert_eq!(pub0.space(), 4);
	}
	}