Created multiplexing publish / subscribe example.

node/src/lib.rs

@@ -1,7 +1,7 @@
 #![no_std]
 #![feature(async_fn_in_trait)]
 
-mod transducer;
+pub mod transducer;
 
 pub mod cell {
     pub use physical::cell::*;

node/src/transducer/input.rs

@@ -1,30 +1,105 @@
 use crate::cell::CellView;
+use crate::transducer::Publisher;
+use core::cell::Cell;
 #[cfg(feature = "embassy-sync")]
 use embassy_sync::blocking_mutex::raw::RawMutex;
 #[cfg(feature = "embassy-sync")]
 use embassy_sync::pubsub::PubSubChannel;
+use embassy_sync::pubsub::{Error, PubSubBehavior, Subscriber};
 pub use physical::transducer::input::*;
+use physical::transducer::Stateful;
 
 #[cfg(feature = "embassy-sync")]
-pub struct PublishInput<
+pub struct RawPublishInput<T: Copy, MutexT: RawMutex, const CAPACITY: usize, const NUM_SUBS: usize>
+{
+    pub channel: PubSubChannel<MutexT, T, CAPACITY, NUM_SUBS, 0>,
+}
+
+impl<T: Copy, MutexT: RawMutex, const CAPACITY: usize, const NUM_SUBS: usize>
+    RawPublishInput<T, MutexT, CAPACITY, NUM_SUBS>
+{
+    #[inline(always)]
+    pub fn new(channel: PubSubChannel<MutexT, T, CAPACITY, NUM_SUBS, 0>) -> Self {
+        Self { channel }
+    }
+
+    #[inline(always)]
+    pub fn update(&self, value: T) {
+        self.channel.publish_immediate(value);
+    }
+}
+
+impl<T: Copy, MutexT: RawMutex, const CAPACITY: usize, const NUM_SUBS: usize>
+    Publisher<CAPACITY, NUM_SUBS> for RawPublishInput<T, MutexT, CAPACITY, NUM_SUBS>
+{
+    type Value = T;
+    type Mutex = MutexT;
+
+    #[inline(always)]
+    fn subscribe(
+        &self,
+    ) -> Result<Subscriber<Self::Mutex, Self::Value, CAPACITY, NUM_SUBS, 0>, Error> {
+        self.channel.subscriber()
+    }
+}
+
+#[cfg(feature = "embassy-sync")]
+pub struct RawStatePubInput<
     T: Copy,
     MutexT: RawMutex,
     const CAPACITY: usize,
     const NUM_SUBS: usize,
-    const NUM_PUBS: usize,
 > {
-    channel: PubSubChannel<MutexT, T, CAPACITY, NUM_SUBS, NUM_PUBS>,
+    pub state_cell: Cell<T>,
+    pub channel: PubSubChannel<MutexT, T, CAPACITY, NUM_SUBS, 0>,
 }
 
-#[cfg(feature = "embassy-sync")]
-pub struct StatefulPublishInput<
-    'a,
-    T: Copy,
-    MutexT: RawMutex,
-    const CAPACITY: usize,
-    const NUM_SUBS: usize,
-    const NUM_PUBS: usize,
-> {
-    pub state_cell: CellView<'a, T>,
-    channel: PubSubChannel<MutexT, T, CAPACITY, NUM_SUBS, NUM_PUBS>,
+impl<T: Copy, MutexT: RawMutex, const CAPACITY: usize, const NUM_SUBS: usize>
+    RawStatePubInput<T, MutexT, CAPACITY, NUM_SUBS>
+{
+    #[inline(always)]
+    pub fn new(
+        state_cell: Cell<T>,
+        channel: PubSubChannel<MutexT, T, CAPACITY, NUM_SUBS, 0>,
+    ) -> Self {
+        Self {
+            state_cell,
+            channel,
+        }
+    }
+
+    #[inline]
+    pub fn update(&self, value: T) {
+        self.state_cell.set(value);
+        self.channel.publish_immediate(value);
+    }
+}
+
+impl<T: Copy, MutexT: RawMutex, const CAPACITY: usize, const NUM_SUBS: usize> Stateful
+    for RawStatePubInput<T, MutexT, CAPACITY, NUM_SUBS>
+{
+    type Value = T;
+
+    #[inline(always)]
+    fn state_cell(&self) -> CellView<Self::Value> {
+        (&self.state_cell).into()
+    }
+
+    #[inline(always)]
+    fn state(&self) -> Self::Value {
+        self.state_cell.get()
+    }
+}
+
+impl<T: Copy, MutexT: RawMutex, const CAPACITY: usize, const NUM_SUBS: usize>
+    Publisher<CAPACITY, NUM_SUBS> for RawStatePubInput<T, MutexT, CAPACITY, NUM_SUBS>
+{
+    type Value = T;
+    type Mutex = MutexT;
+
+    fn subscribe(
+        &self,
+    ) -> Result<Subscriber<Self::Mutex, Self::Value, CAPACITY, NUM_SUBS, 0>, Error> {
+        self.channel.subscriber()
+    }
 }

node/src/transducer/mod.rs

@@ -1,5 +1,5 @@
-mod input;
-mod output;
+pub mod input;
+pub mod output;
 
 pub use physical::transducer::*;
 
@@ -11,17 +11,11 @@ use embassy_sync::pubsub;
 use embassy_sync::pubsub::Subscriber;
 
 #[cfg(feature = "embassy-sync")]
-pub trait Publisher {
+pub trait Publisher<const CAPACITY: usize, const NUM_SUBS: usize> {
     type Value: Copy;
     type Mutex: RawMutex;
-    const CAPACITY: usize;
-    const NUM_SUBS: usize;
-    const NUM_PUBS: usize;
 
     fn subscribe(
         &self,
-    ) -> Result<
-        Subscriber<Self::Mutex, Self::Value, Self::CAPACITY, Self::NUM_SUBS, Self::NUM_PUBS>,
-        pubsub::Error,
-    >;
+    ) -> Result<Subscriber<Self::Mutex, Self::Value, CAPACITY, NUM_SUBS, 0>, pubsub::Error>;
 }