Add relayable service event plumbing

battery-service-relay/src/lib.rs

@@ -23,6 +23,9 @@ impl<S: battery_service_interface::BatteryService> embedded_services::relay::mct
 {
     type RequestType = serialization::AcpiBatteryRequest;
     type ResultType = serialization::AcpiBatteryResult;
+
+    // Temporary until figure out what events want to send
+    type EventType = ();
 }
 
 impl<S: battery_service_interface::BatteryService> embedded_services::relay::mctp::RelayServiceHandler

debug-service/src/debug_service.rs

@@ -37,6 +37,9 @@ impl Service {
 impl embedded_services::relay::mctp::RelayServiceHandlerTypes for Service {
     type RequestType = DebugRequest;
     type ResultType = DebugResult;
+
+    // Temporary until figure out what events want to send
+    type EventType = ();
 }
 
 impl embedded_services::relay::mctp::RelayServiceHandler for Service {

embedded-service/src/event.rs

@@ -143,3 +143,157 @@ impl<I, O, S: Sender<O>, F: FnMut(I) -> O> Sender<I> for MapSender<I, O, S, F> {
         self.sender.send((self.map_fn)(event))
     }
 }
+
+/// Applies a function on events received from the wrapped receiver.
+pub struct MapReceiver<I, O, R: Receiver<I>, F: FnMut(I) -> O> {
+    receiver: R,
+    map_fn: F,
+    _phantom: PhantomData<(I, O)>,
+}
+
+impl<I, O, R: Receiver<I>, F: FnMut(I) -> O> MapReceiver<I, O, R, F> {
+    /// Create a new MapReceiver.
+    pub fn new(receiver: R, map_fn: F) -> Self {
+        Self {
+            receiver,
+            map_fn,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<I, O, R: Receiver<I>, F: FnMut(I) -> O> Receiver<O> for MapReceiver<I, O, R, F> {
+    fn try_next(&mut self) -> Option<O> {
+        self.receiver.try_next().map(&mut self.map_fn)
+    }
+
+    async fn wait_next(&mut self) -> O {
+        (self.map_fn)(self.receiver.wait_next().await)
+    }
+}
+
+/// Filters events from the wrapped receiver, only yielding events that pass the predicate.
+///
+/// Events that do not pass the filter are consumed and discarded.
+pub struct FilterReceiver<E, R: Receiver<E>, F: FnMut(&E) -> bool> {
+    receiver: R,
+    filter_fn: F,
+    _phantom: PhantomData<E>,
+}
+
+impl<E, R: Receiver<E>, F: FnMut(&E) -> bool> FilterReceiver<E, R, F> {
+    /// Create a new FilterReceiver.
+    pub fn new(receiver: R, filter_fn: F) -> Self {
+        Self {
+            receiver,
+            filter_fn,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<E, R: Receiver<E>, F: FnMut(&E) -> bool> Receiver<E> for FilterReceiver<E, R, F> {
+    fn try_next(&mut self) -> Option<E> {
+        loop {
+            match self.receiver.try_next() {
+                Some(e) if (self.filter_fn)(&e) => return Some(e),
+                Some(_) => continue,
+                None => return None,
+            }
+        }
+    }
+
+    async fn wait_next(&mut self) -> E {
+        loop {
+            let e = self.receiver.wait_next().await;
+            if (self.filter_fn)(&e) {
+                return e;
+            }
+        }
+    }
+}
+
+/// A receiver that never produces events.
+///
+/// This is mainly used to make it easier to construct a `MuxReceiver`
+/// via macro since we don't need to handle the special start case
+/// when chaining `with` calls.
+pub struct NeverReceiver<E>(PhantomData<E>);
+
+impl<E> NeverReceiver<E> {
+    /// Create a new NeverReceiver.
+    pub fn new() -> Self {
+        Self(PhantomData)
+    }
+}
+
+impl<E> Default for NeverReceiver<E> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<E> Receiver<E> for NeverReceiver<E> {
+    fn try_next(&mut self) -> Option<E> {
+        None
+    }
+
+    async fn wait_next(&mut self) -> E {
+        core::future::pending().await
+    }
+}
+
+/// Combines multiple receivers into one by racing them (with left-bias) and returning
+/// the first event that becomes available mapped to a common event type.
+pub struct MuxReceiver<E, L: Receiver<E>, R: Receiver<E>> {
+    left: L,
+    right: R,
+    _phantom: PhantomData<E>,
+}
+
+impl<E> MuxReceiver<E, NeverReceiver<E>, NeverReceiver<E>> {
+    /// Create an empty MuxReceiver.
+    ///
+    /// Use `.with()` to add receivers.
+    pub fn new() -> Self {
+        Self {
+            left: NeverReceiver::new(),
+            right: NeverReceiver::new(),
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<E> Default for MuxReceiver<E, NeverReceiver<E>, NeverReceiver<E>> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<E, L: Receiver<E>, R1: Receiver<E>> MuxReceiver<E, L, R1> {
+    /// Add another receiver to multiplex with this one.
+    pub fn with<I, R2: Receiver<I>, F: FnMut(I) -> E>(
+        self,
+        receiver: R2,
+        map_fn: F,
+    ) -> MuxReceiver<E, Self, MapReceiver<I, E, R2, F>> {
+        MuxReceiver {
+            left: self,
+            right: MapReceiver::new(receiver, map_fn),
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<E, L: Receiver<E>, R: Receiver<E>> Receiver<E> for MuxReceiver<E, L, R> {
+    fn try_next(&mut self) -> Option<E> {
+        self.left.try_next().or_else(|| self.right.try_next())
+    }
+
+    async fn wait_next(&mut self) -> E {
+        match embassy_futures::select::select(self.left.wait_next(), self.right.wait_next()).await {
+            embassy_futures::select::Either::First(e) => e,
+            embassy_futures::select::Either::Second(e) => e,
+        }
+    }
+}

embedded-service/src/relay/mod.rs

@@ -116,6 +116,9 @@ pub mod mctp {
 
         /// The result type that this service handler processes
         type ResultType: super::SerializableResult;
+
+        /// The event type that this service emits.
+        type EventType;
     }
 
     /// Trait for a service that can be relayed over an external bus (e.g. battery service, thermal service, time-alarm service)
@@ -448,6 +451,14 @@ pub mod mctp {
                     }
 
 
+                    /// A common event type wrapper for all relayable service events.
+                    #[derive(Debug)]
+                    pub enum ServiceEvent {
+                        $(
+                            $service_name(<$service_handler_type as $crate::relay::mctp::RelayServiceHandlerTypes>::EventType),
+                        )+
+                    }
+
                     pub struct $relay_type_name {
                         $(
                             [<$service_name:snake _handler>]: $service_handler_type,
@@ -466,6 +477,33 @@ pub mod mctp {
                                 )+
                             }
                         }
+
+                        /// Build an event multiplexer from the provided relayable services.
+                        ///
+                        /// This is generic over the receiver type used for each service, so callers
+                        /// can decide at the call site which concrete `Receiver<EventType>` impl to
+                        /// supply for each relayed service (e.g. `NeverReceiver`, an embassy channel
+                        /// `Receiver`, a `DynamicReceiver`, or a custom impl).
+                        ///
+                        /// The caller will then need to further filter this event mux to whatever
+                        /// events they consider worth notifying the host about.
+                        ///
+                        /// Lastly, the caller will then need to map the events into a format the
+                        /// relay service understands (e.g. a single u8 for uart-service).
+                        pub fn event_mux<
+                            $(
+                                [<$service_name Rx>]: $crate::event::Receiver<
+                                    <$service_handler_type as $crate::relay::mctp::RelayServiceHandlerTypes>::EventType
+                                >,
+                            )+
+                        >(
+                            $(
+                                [<$service_name:snake _event_rx>]: [<$service_name Rx>],
+                            )+
+                        ) -> impl $crate::event::Receiver<ServiceEvent> {
+                            $crate::event::MuxReceiver::new()
+                                $(.with([<$service_name:snake _event_rx>], ServiceEvent::$service_name))+
+                        }
                     }
 
                     impl $crate::relay::mctp::RelayHandler for $relay_type_name {
@@ -494,6 +532,7 @@ pub mod mctp {
 
                 // Allows this generated relay type to be publicly re-exported
                 pub use [< _odp_impl_ $relay_type_name:snake >]::$relay_type_name;
+                pub use [< _odp_impl_ $relay_type_name:snake >]::ServiceEvent;
 
             } // end paste!
         }; // end macro arm

thermal-service-interface/src/lib.rs

@@ -3,6 +3,17 @@
 pub mod fan;
 pub mod sensor;
 
+/// Thermal service event.
+#[derive(Debug, PartialEq, Clone, Copy)]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
+#[non_exhaustive]
+pub enum Event {
+    /// A sensor event occurred.
+    Sensor(u8, sensor::Event),
+    /// A fan event occurred.
+    Fan(u8, fan::Event),
+}
+
 /// Thermal service interface trait.
 pub trait ThermalService {
     /// Associated type for registered sensor services.

thermal-service-relay/src/lib.rs

@@ -3,6 +3,7 @@
 mod serialization;
 
 pub use serialization::{ThermalError, ThermalRequest, ThermalResponse, ThermalResult};
+use thermal_service_interface::Event as ThermalEvent;
 use thermal_service_interface::ThermalService;
 use thermal_service_interface::fan::{self, FanService};
 use thermal_service_interface::sensor::{self, SensorService};
@@ -194,6 +195,7 @@ impl<T: ThermalService> ThermalServiceRelayHandler<T> {
 impl<T: ThermalService> embedded_services::relay::mctp::RelayServiceHandlerTypes for ThermalServiceRelayHandler<T> {
     type RequestType = ThermalRequest;
     type ResultType = ThermalResult;
+    type EventType = ThermalEvent;
 }
 
 impl<T: ThermalService> embedded_services::relay::mctp::RelayServiceHandler for ThermalServiceRelayHandler<T> {

time-alarm-service-relay/src/lib.rs

@@ -20,6 +20,9 @@ impl<T: TimeAlarmService> TimeAlarmServiceRelayHandler<T> {
 impl<T: TimeAlarmService> embedded_services::relay::mctp::RelayServiceHandlerTypes for TimeAlarmServiceRelayHandler<T> {
     type RequestType = AcpiTimeAlarmRequest;
     type ResultType = AcpiTimeAlarmResult;
+
+    // Temporary until figure out what events want to send
+    type EventType = ();
 }
 
 impl<T: TimeAlarmService> embedded_services::relay::mctp::RelayServiceHandler for TimeAlarmServiceRelayHandler<T> {

uart-service/Cargo.toml

@@ -17,6 +17,7 @@ workspace = true
 embedded-services.workspace = true
 defmt = { workspace = true, optional = true }
 log = { workspace = true, optional = true }
+embassy-futures.workspace = true
 embassy-sync.workspace = true
 mctp-rs = { workspace = true }
 embedded-io-async.workspace = true