[RFC] Relay service event handling

battery-service-relay/src/lib.rs

@@ -23,6 +23,10 @@ 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 = core::convert::Infallible;
+    type EventReceiver = embedded_services::event::NeverReceiver<core::convert::Infallible>;
 }
 
 impl<S: battery_service_interface::BatteryService> embedded_services::relay::mctp::RelayServiceHandler

debug-service/src/debug_service.rs

@@ -37,6 +37,10 @@ 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 = core::convert::Infallible;
+    type EventReceiver = embedded_services::event::NeverReceiver<core::convert::Infallible>;
 }
 
 impl embedded_services::relay::mctp::RelayServiceHandler for Service {

embedded-service/src/event.rs

@@ -143,3 +143,116 @@ 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)
+    }
+}
+
+/// 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 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,12 @@ pub mod mctp {
 
         /// The result type that this service handler processes
         type ResultType: super::SerializableResult;
+
+        /// The event type that this service emits.
+        type EventType;
+
+        /// The receiver type used to receive events from this service.
+        type EventReceiver: crate::event::Receiver<Self::EventType>;
     }
 
     /// Trait for a service that can be relayed over an external bus (e.g. battery service, thermal service, time-alarm service)
@@ -448,6 +454,13 @@ pub mod mctp {
                     }
 
 
+                    /// A common event type wrapper for all relayable service events.
+                    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 +479,23 @@ pub mod mctp {
                                 )+
                             }
                         }
+
+                        /// Build an event multiplexer from the provided relayable services.
+                        ///
+                        /// This will be constructed similar to the relay handler,
+                        /// except we pass in the event receivers for each service.
+                        ///
+                        /// My idea is this can then be passed into a relay service (again, like the relay handler),
+                        /// but this poses a problem: the relay service doesn't know what the macro'd
+                        /// `ServiceEvent` actually looks like...
+                        pub fn event_mux(
+                            $(
+                                [<$service_name:snake _event_rx>]: <$service_handler_type as $crate::relay::mctp::RelayServiceHandlerTypes>::EventReceiver,
+                            )+
+                        ) -> 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 +524,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

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

@@ -53,6 +53,9 @@ async fn main(spawner: embassy_executor::Spawner) {
 
     let _relay_handler = EspiRelayHandler::new(TimeAlarmServiceRelayHandlerType::new(time_service));
 
+    // Temporary example of constructing the event mux
+    let mut _event_mux = EspiRelayHandler::event_mux(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-relay/src/lib.rs

@@ -194,6 +194,10 @@ impl<T: ThermalService> ThermalServiceRelayHandler<T> {
 impl<T: ThermalService> embedded_services::relay::mctp::RelayServiceHandlerTypes for ThermalServiceRelayHandler<T> {
     type RequestType = ThermalRequest;
     type ResultType = ThermalResult;
+
+    /// Temporary until figure out what events want to send.
+    type EventType = core::convert::Infallible;
+    type EventReceiver = embedded_services::event::NeverReceiver<core::convert::Infallible>;
 }
 
 impl<T: ThermalService> embedded_services::relay::mctp::RelayServiceHandler for ThermalServiceRelayHandler<T> {

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

@@ -20,6 +20,10 @@ 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 = core::convert::Infallible;
+    type EventReceiver = embedded_services::event::NeverReceiver<core::convert::Infallible>;
 }
 
 impl<T: TimeAlarmService> embedded_services::relay::mctp::RelayServiceHandler for TimeAlarmServiceRelayHandler<T> {