1 files changed, 67 insertions, 0 deletions

diff --git a/embassy-sync/src/pubsub/publisher.rs b/embassy-sync/src/pubsub/publisher.rs
index e66b3b1db..7a1ab66de 100644
--- a/embassy-sync/src/pubsub/publisher.rs
+++ b/embassy-sync/src/pubsub/publisher.rs
@@ -74,6 +74,12 @@ impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Pub<'a, PSB, T> {
     pub fn is_full(&self) -> bool {
         self.channel.is_full()
     }
+
+    /// Create a [`futures::Sink`] adapter for this publisher.
+    #[inline]
+    pub const fn sink(&self) -> PubSink<'a, '_, PSB, T> {
+        PubSink { publ: self, fut: None }
+    }
 }
 
 impl<'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> Drop for Pub<'a, PSB, T> {
@@ -221,6 +227,67 @@ impl<'a, M: RawMutex, T: Clone, const CAP: usize, const SUBS: usize, const PUBS:
     }
 }
 
+#[must_use = "Sinks do nothing unless polled"]
+/// [`futures_sink::Sink`] adapter for [`Pub`].
+pub struct PubSink<'a, 'p, PSB, T>
+where
+    T: Clone,
+    PSB: PubSubBehavior<T> + ?Sized,
+{
+    publ: &'p Pub<'a, PSB, T>,
+    fut: Option<PublisherWaitFuture<'p, 'a, PSB, T>>,
+}
+
+impl<'a, 'p, PSB, T> PubSink<'a, 'p, PSB, T>
+where
+    PSB: PubSubBehavior<T> + ?Sized,
+    T: Clone,
+{
+    /// Try to make progress on the pending future if we have one.
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
+        let Some(mut fut) = self.fut.take() else {
+            return Poll::Ready(());
+        };
+
+        if Pin::new(&mut fut).poll(cx).is_pending() {
+            self.fut = Some(fut);
+            return Poll::Pending;
+        }
+
+        Poll::Ready(())
+    }
+}
+
+impl<'a, 'p, PSB, T> futures_sink::Sink<T> for PubSink<'a, 'p, PSB, T>
+where
+    PSB: PubSubBehavior<T> + ?Sized,
+    T: Clone,
+{
+    type Error = core::convert::Infallible;
+
+    #[inline]
+    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.poll(cx).map(Ok)
+    }
+
+    #[inline]
+    fn start_send(mut self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
+        self.fut = Some(self.publ.publish(item));
+
+        Ok(())
+    }
+
+    #[inline]
+    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.poll(cx).map(Ok)
+    }
+
+    #[inline]
+    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.poll(cx).map(Ok)
+    }
+}
+
 /// Future for the publisher wait action
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct PublisherWaitFuture<'s, 'a, PSB: PubSubBehavior<T> + ?Sized, T: Clone> {