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

@@ -32,6 +32,11 @@ pub trait Receiver<E> {
     /// Return none if there are no pending events
     fn try_next(&mut self) -> Option<E>;
     /// Receive an event
+    ///
+    /// # Cancel Safety
+    ///
+    /// The implementation MUST be cancel-safe as the `wait_next` method
+    /// is typically called inside a `select!` and thus may be cancelled.
     fn wait_next(&mut self) -> impl Future<Output = E>;
 }
 
@@ -191,3 +196,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)
@@ -169,6 +172,11 @@ pub mod mctp {
             &'a self,
             message: Self::RequestEnumType,
         ) -> impl core::future::Future<Output = Self::ResultEnumType> + 'a;
+
+        /// Returns a mutable reference to the multiplexed receiver of all relayable service events.
+        ///
+        /// Each service's events are mapped to the service's u8 ID.
+        fn receiver(&mut self) -> &mut impl crate::event::Receiver<u8>;
     }
 
     /// This macro generates a relay type over a collection of message types, which can be used by a relay service to
@@ -448,27 +456,42 @@ pub mod mctp {
                     }
 
 
-                    pub struct $relay_type_name {
+                    pub struct $relay_type_name<T: $crate::event::Receiver<u8>> {
                         $(
                             [<$service_name:snake _handler>]: $service_handler_type,
                         )+
+                        receiver: T,
                     }
 
-                    impl $relay_type_name {
-                        pub fn new(
+                    // Note: Need a concrete type here to satisfy the type system,
+                    // so we just use `NeverReceiver` as a placeholder
+                    impl $relay_type_name<$crate::event::NeverReceiver<u8>> {
+                        pub fn new<
+                            $(
+                                [<$service_name Rx>]: $crate::event::Receiver<
+                                    <$service_handler_type as $crate::relay::mctp::RelayServiceHandlerTypes>::EventType
+                                >,
+                            )+
+                        >(
                             $(
                                 [<$service_name:snake _handler>]: $service_handler_type,
                             )+
-                        ) -> Self {
-                            Self {
+                            $(
+                                [<$service_name:snake _event_rx>]: [<$service_name Rx>],
+                            )+
+                        ) -> $relay_type_name<impl $crate::event::Receiver<u8>> {
+                            let receiver = $crate::event::MuxReceiver::new()
+                                $(.with([<$service_name:snake _event_rx>], |_| $service_id as u8))+;
+                            $relay_type_name {
                                 $(
                                     [<$service_name:snake _handler>],
                                 )+
+                                receiver,
                             }
                         }
                     }
 
-                    impl $crate::relay::mctp::RelayHandler for $relay_type_name {
+                    impl<T: $crate::event::Receiver<u8>> $crate::relay::mctp::RelayHandler for $relay_type_name<T> {
                         type ServiceIdType = OdpService;
                         type HeaderType = OdpHeader;
                         type RequestEnumType = HostRequest;
@@ -489,6 +512,10 @@ pub mod mctp {
                                 }
                             }
                         }
+
+                        fn receiver(&mut self) -> &mut impl $crate::event::Receiver<u8> {
+                            &mut self.receiver
+                        }
                     }
                 } // end mod __odp_impl
 

examples/rt685s-evk/src/bin/time_alarm.rs

@@ -51,7 +51,10 @@ async fn main(spawner: embassy_executor::Spawner) {
         TimeAlarm, 0x0B, crate::TimeAlarmServiceRelayHandlerType;
     );
 
-    let _relay_handler = EspiRelayHandler::new(TimeAlarmServiceRelayHandlerType::new(time_service));
+    let _relay_handler = EspiRelayHandler::new(
+        TimeAlarmServiceRelayHandlerType::new(time_service),
+        embedded_services::event::NeverReceiver::new(),
+    );
 
     // Here, you'd normally pass _relay_handler to your relay service (e.g. eSPI service).
     // In this example, we're not leveraging a relay service, so we'll just demonstrate some direct calls.

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