Futures in pub & sub are now awaited instead of returned for better user compiler diagnostics.

Added functions for reading how many messages are available

3 files changed, 92 insertions, 5 deletions

diff --git a/embassy-sync/src/pubsub/mod.rs b/embassy-sync/src/pubsub/mod.rs
index 62a9e4763..335d7e33e 100644
--- a/embassy-sync/src/pubsub/mod.rs
+++ b/embassy-sync/src/pubsub/mod.rs
@@ -192,6 +192,10 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
         })
     }
 
+    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();
@@ -217,6 +221,13 @@ impl<M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS: usi
         })
     }
 
+    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();
@@ -388,6 +399,10 @@ pub trait PubSubBehavior<T> {
     /// 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.
@@ -396,6 +411,9 @@ pub trait PubSubBehavior<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);
 
@@ -539,4 +557,59 @@ mod tests {
 
         drop(sub0);
     }
+
+    #[futures_test::test]
+    async fn correct_available() {
+        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!(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);
+    }
 }
diff --git a/embassy-sync/src/pubsub/publisher.rs b/embassy-sync/src/pubsub/publisher.rs
index 705797f60..484f1dbfd 100644
--- a/embassy-sync/src/pubsub/publisher.rs
+++ b/embassy-sync/src/pubsub/publisher.rs
@@ -31,17 +31,26 @@ impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Pub<'a, PSB, T> {
     }
 
     /// Publish a message. But if the message queue is full, wait for all subscribers to have read the last message
-    pub fn publish<'s>(&'s self, message: T) -> PublisherWaitFuture<'s, 'a, PSB, T> {
+    pub async fn publish<'s>(&'s self, message: T) {
         PublisherWaitFuture {
             message: Some(message),
             publisher: self,
         }
+        .await
     }
 
     /// Publish a message if there is space in the message queue
     pub fn try_publish(&self, message: T) -> Result<(), T> {
         self.channel.publish_with_context(message, None)
     }
+
+    /// The amount of messages that can still be published without having to wait or without having to lag the subscribers
+    ///
+    /// *Note: In the time between checking this and a publish action, other publishers may have had time to publish something.
+    /// So checking doesn't give any guarantees.*
+    pub fn space(&self) -> usize {
+        self.channel.space()
+    }
 }
 
 impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Drop for Pub<'a, PSB, T> {
@@ -158,7 +167,7 @@ impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS:
 }
 
 /// Future for the publisher wait action
-pub struct PublisherWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
+struct PublisherWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
     /// The message we need to publish
     message: Option<T>,
     publisher: &'s Pub<'a, PSB, T>,
diff --git a/embassy-sync/src/pubsub/subscriber.rs b/embassy-sync/src/pubsub/subscriber.rs
index b9a2cbe18..8a8e9144b 100644
--- a/embassy-sync/src/pubsub/subscriber.rs
+++ b/embassy-sync/src/pubsub/subscriber.rs
@@ -28,8 +28,8 @@ impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Sub<'a, PSB, T> {
     }
 
     /// Wait for a published message
-    pub fn next_message<'s>(&'s mut self) -> SubscriberWaitFuture<'s, 'a, PSB, T> {
-        SubscriberWaitFuture { subscriber: self }
+    pub async fn next_message(&mut self) -> WaitResult<T> {
+        SubscriberWaitFuture { subscriber: self }.await
     }
 
     /// Wait for a published message (ignoring lag results)
@@ -64,6 +64,11 @@ impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Sub<'a, PSB, T> {
             }
         }
     }
+
+    /// The amount of messages this subscriber hasn't received yet
+    pub fn available(&self) -> u64 {
+        self.channel.available(self.next_message_id)
+    }
 }
 
 impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Drop for Sub<'a, PSB, T> {
@@ -135,7 +140,7 @@ impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS:
 }
 
 /// Future for the subscriber wait action
-pub struct SubscriberWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
+struct SubscriberWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {
     subscriber: &'s mut Sub<'a, PSB, T>,
 }