Merge pull request #3797 from stargazing-dino/add-channel-stream

Add stream impl for embassy-sync Channels

1 files changed, 22 insertions, 0 deletions

diff --git a/embassy-sync/src/channel.rs b/embassy-sync/src/channel.rs
index 9a7d2fa2f..4d1fa9e39 100644
--- a/embassy-sync/src/channel.rs
+++ b/embassy-sync/src/channel.rs
@@ -317,6 +317,17 @@ where
     }
 }
 
+impl<'ch, M, T, const N: usize> futures_util::Stream for Receiver<'ch, M, T, N>
+where
+    M: RawMutex,
+{
+    type Item = T;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.channel.poll_receive(cx).map(Some)
+    }
+}
+
 /// Future returned by [`Channel::receive`] and  [`Receiver::receive`].
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct ReceiveFuture<'ch, M, T, const N: usize>
@@ -771,6 +782,17 @@ where
     }
 }
 
+impl<M, T, const N: usize> futures_util::Stream for Channel<M, T, N>
+where
+    M: RawMutex,
+{
+    type Item = T;
+
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.poll_receive(cx).map(Some)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use core::time::Duration;