Power policy event pass and clean-up

embedded-service/src/event.rs

@@ -1,5 +1,7 @@
 //! Common traits for event senders and receivers
 
+use core::marker::PhantomData;
+
 use embassy_sync::channel::{DynamicReceiver, DynamicSender};
 
 /// Common event sender trait
@@ -41,3 +43,42 @@ impl<E> Receiver<E> for DynamicReceiver<'_, E> {
         self.receive()
     }
 }
+
+/// A sender that discards all events sent to it.
+pub struct NoopSender;
+
+impl<E> Sender<E> for NoopSender {
+    fn try_send(&mut self, _event: E) -> Option<()> {
+        Some(())
+    }
+
+    async fn send(&mut self, _event: E) {}
+}
+
+/// Applies a function on events before passing them to the wrapped sender
+pub struct MapSender<I, O, S: Sender<O>, F: FnMut(I) -> O> {
+    sender: S,
+    map_fn: F,
+    _phantom: PhantomData<(I, O)>,
+}
+
+impl<I, O, S: Sender<O>, F: FnMut(I) -> O> MapSender<I, O, S, F> {
+    /// Create a new self
+    pub fn new(sender: S, map_fn: F) -> Self {
+        Self {
+            sender,
+            map_fn,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<I, O, S: Sender<O>, F: FnMut(I) -> O> Sender<I> for MapSender<I, O, S, F> {
+    fn try_send(&mut self, event: I) -> Option<()> {
+        self.sender.try_send((self.map_fn)(event))
+    }
+
+    fn send(&mut self, event: I) -> impl Future<Output = ()> {
+        self.sender.send((self.map_fn)(event))
+    }
+}

embedded-service/src/lib.rs

@@ -21,6 +21,7 @@ pub mod hid;
 pub mod init;
 pub mod ipc;
 pub mod keyboard;
+pub mod named;
 pub mod relay;
 pub mod sync;
 

embedded-service/src/named.rs

@@ -0,0 +1,7 @@
+//! Traits for things that have names.
+
+/// Trait for anything that has a name.
+pub trait Named {
+    /// Return name
+    fn name(&self) -> &'static str;
+}

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

@@ -15,11 +15,13 @@ use embassy_sync::once_lock::OnceLock;
 use embassy_sync::pubsub::PubSubChannel;
 use embassy_time::{self as _, Delay};
 use embedded_cfu_protocol::protocol_definitions::{FwUpdateOffer, FwUpdateOfferResponse, FwVersion, HostToken};
+use embedded_services::event::NoopSender;
 use embedded_services::{GlobalRawMutex, IntrusiveList};
 use embedded_services::{error, info};
 use embedded_usb_pd::GlobalPortId;
 use power_policy_interface::psu;
-use power_policy_service::psu::EventReceivers;
+use power_policy_service::psu::ArrayEventReceivers;
+use power_policy_service::service::registration::ArrayRegistration;
 use static_cell::StaticCell;
 use tps6699x::asynchronous::embassy as tps6699x;
 use type_c_service::driver::tps6699x::{self as tps6699x_drv};
@@ -64,6 +66,11 @@ type Wrapper<'a> = ControllerWrapper<
 type Controller<'a> = tps6699x::controller::Controller<GlobalRawMutex, BusDevice<'a>>;
 type Interrupt<'a> = tps6699x::Interrupt<'a, GlobalRawMutex, BusDevice<'a>>;
 
+type PowerPolicyServiceType = Mutex<
+    GlobalRawMutex,
+    power_policy_service::service::Service<'static, ArrayRegistration<'static, DeviceType, 2, NoopSender, 1>>,
+>;
+
 #[embassy_executor::task]
 async fn pd_controller_task(controller: &'static Wrapper<'static>) {
     loop {
@@ -80,8 +87,8 @@ async fn interrupt_task(mut int_in: Input<'static>, mut interrupt: Interrupt<'st
 
 #[embassy_executor::task]
 async fn power_policy_task(
-    psu_events: EventReceivers<'static, 2, DeviceType, DynamicReceiver<'static, psu::event::EventData>>,
-    power_policy: &'static Mutex<GlobalRawMutex, power_policy_service::service::Service<'static, DeviceType>>,
+    psu_events: ArrayEventReceivers<'static, 2, DeviceType, DynamicReceiver<'static, psu::event::EventData>>,
+    power_policy: &'static PowerPolicyServiceType,
 ) {
     power_policy_service::service::task::task(psu_events, power_policy).await;
 }
@@ -204,13 +211,14 @@ async fn main(spawner: Spawner) {
     static POWER_SERVICE_CONTEXT: StaticCell<power_policy_service::service::context::Context> = StaticCell::new();
     let power_service_context = POWER_SERVICE_CONTEXT.init(power_policy_service::service::context::Context::new());
 
-    static POWER_POLICY_PSU_REGISTRATION: StaticCell<[&DeviceType; 2]> = StaticCell::new();
-    let psu_registration = POWER_POLICY_PSU_REGISTRATION.init([&wrapper.ports[0].proxy, &wrapper.ports[1].proxy]);
+    let power_policy_registration = ArrayRegistration {
+        psus: [&wrapper.ports[0].proxy, &wrapper.ports[1].proxy],
+        service_senders: [NoopSender],
+    };
 
-    static POWER_SERVICE: StaticCell<Mutex<GlobalRawMutex, power_policy_service::service::Service<DeviceType>>> =
-        StaticCell::new();
+    static POWER_SERVICE: StaticCell<PowerPolicyServiceType> = StaticCell::new();
     let power_service = POWER_SERVICE.init(Mutex::new(power_policy_service::service::Service::new(
-        psu_registration,
+        power_policy_registration,
         power_service_context,
         power_policy_service::service::config::Config::default(),
     )));
@@ -233,7 +241,7 @@ async fn main(spawner: Spawner) {
 
     info!("Spawining power policy task");
     spawner.must_spawn(power_policy_task(
-        EventReceivers::new(
+        ArrayEventReceivers::new(
             [&wrapper.ports[0].proxy, &wrapper.ports[1].proxy],
             [policy_receiver0, policy_receiver1],
         ),

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

@@ -18,11 +18,13 @@ use embassy_time::Timer;
 use embassy_time::{self as _, Delay};
 use embedded_cfu_protocol::protocol_definitions::*;
 use embedded_cfu_protocol::protocol_definitions::{FwUpdateOffer, FwUpdateOfferResponse, FwVersion};
+use embedded_services::event::NoopSender;
 use embedded_services::{GlobalRawMutex, IntrusiveList};
 use embedded_services::{error, info};
 use embedded_usb_pd::GlobalPortId;
 use power_policy_interface::psu;
-use power_policy_service::psu::EventReceivers;
+use power_policy_service::psu::ArrayEventReceivers;
+use power_policy_service::service::registration::ArrayRegistration;
 use static_cell::StaticCell;
 use tps6699x::asynchronous::embassy as tps6699x;
 use type_c_service::driver::tps6699x::{self as tps6699x_drv};
@@ -62,6 +64,11 @@ type Wrapper<'a> = ControllerWrapper<
 type Controller<'a> = tps6699x::controller::Controller<GlobalRawMutex, BusDevice<'a>>;
 type Interrupt<'a> = tps6699x::Interrupt<'a, GlobalRawMutex, BusDevice<'a>>;
 
+type PowerPolicyServiceType = Mutex<
+    GlobalRawMutex,
+    power_policy_service::service::Service<'static, ArrayRegistration<'static, DeviceType, 2, NoopSender, 1>>,
+>;
+
 const NUM_PD_CONTROLLERS: usize = 1;
 const CONTROLLER0_ID: ControllerId = ControllerId(0);
 const CONTROLLER0_CFU_ID: ComponentId = 0x12;
@@ -164,8 +171,8 @@ async fn fw_update_task() {
 
 #[embassy_executor::task]
 async fn power_policy_task(
-    psu_events: EventReceivers<'static, 2, DeviceType, DynamicReceiver<'static, psu::event::EventData>>,
-    power_policy: &'static Mutex<GlobalRawMutex, power_policy_service::service::Service<'static, DeviceType>>,
+    psu_events: ArrayEventReceivers<'static, 2, DeviceType, DynamicReceiver<'static, psu::event::EventData>>,
+    power_policy: &'static PowerPolicyServiceType,
 ) {
     power_policy_service::service::task::task(psu_events, power_policy).await;
 }
@@ -282,13 +289,14 @@ async fn main(spawner: Spawner) {
     static POWER_SERVICE_CONTEXT: StaticCell<power_policy_service::service::context::Context> = StaticCell::new();
     let power_service_context = POWER_SERVICE_CONTEXT.init(power_policy_service::service::context::Context::new());
 
-    static POWER_POLICY_PSU_REGISTRATION: StaticCell<[&DeviceType; 2]> = StaticCell::new();
-    let psu_registration = POWER_POLICY_PSU_REGISTRATION.init([&wrapper.ports[0].proxy, &wrapper.ports[1].proxy]);
+    let power_policy_registration = ArrayRegistration {
+        psus: [&wrapper.ports[0].proxy, &wrapper.ports[1].proxy],
+        service_senders: [NoopSender],
+    };
 
-    static POWER_SERVICE: StaticCell<Mutex<GlobalRawMutex, power_policy_service::service::Service<DeviceType>>> =
-        StaticCell::new();
+    static POWER_SERVICE: StaticCell<PowerPolicyServiceType> = StaticCell::new();
     let power_service = POWER_SERVICE.init(Mutex::new(power_policy_service::service::Service::new(
-        psu_registration,
+        power_policy_registration,
         power_service_context,
         power_policy_service::service::config::Config::default(),
     )));
@@ -321,7 +329,7 @@ async fn main(spawner: Spawner) {
 
     info!("Spawining power policy task");
     spawner.must_spawn(power_policy_task(
-        EventReceivers::new(
+        ArrayEventReceivers::new(
             [&wrapper.ports[0].proxy, &wrapper.ports[1].proxy],
             [policy_receiver0, policy_receiver1],
         ),

examples/std/src/bin/power_policy.rs

@@ -5,16 +5,21 @@ use embassy_sync::{
     mutex::Mutex,
 };
 use embassy_time::{self as _, Timer};
-use embedded_services::GlobalRawMutex;
+use embedded_services::{GlobalRawMutex, event::NoopSender, named::Named};
 use log::*;
 use power_policy_interface::psu::{Error, Psu};
 use power_policy_interface::{
     capability::{ConsumerFlags, ConsumerPowerCapability, PowerCapability, ProviderPowerCapability},
     psu,
 };
-use power_policy_service::psu::EventReceivers;
+use power_policy_service::{psu::ArrayEventReceivers, service::registration::ArrayRegistration};
 use static_cell::StaticCell;
 
+type ServiceType = Mutex<
+    GlobalRawMutex,
+    power_policy_service::service::Service<'static, ArrayRegistration<'static, DeviceType, 2, NoopSender, 1>>,
+>;
+
 const LOW_POWER: PowerCapability = PowerCapability {
     voltage_mv: 5000,
     current_ma: 1500,
@@ -96,7 +101,9 @@ impl Psu for ExampleDevice<'_> {
     fn state_mut(&mut self) -> &mut psu::State {
         &mut self.state
     }
+}
 
+impl Named for ExampleDevice<'_> {
     fn name(&self) -> &'static str {
         self.name
     }
@@ -131,19 +138,20 @@ async fn run(spawner: Spawner) {
     static SERVICE_CONTEXT: StaticCell<power_policy_service::service::context::Context> = StaticCell::new();
     let service_context = SERVICE_CONTEXT.init(power_policy_service::service::context::Context::new());
 
-    static POWER_POLICY_PSU_REGISTRATION: StaticCell<[&DeviceType; 2]> = StaticCell::new();
-    let psu_registration = POWER_POLICY_PSU_REGISTRATION.init([device0, device1]);
+    let registration = ArrayRegistration {
+        psus: [device0, device1],
+        service_senders: [NoopSender],
+    };
 
-    static SERVICE: StaticCell<Mutex<GlobalRawMutex, power_policy_service::service::Service<'static, DeviceType>>> =
-        StaticCell::new();
+    static SERVICE: StaticCell<ServiceType> = StaticCell::new();
     let service = SERVICE.init(Mutex::new(power_policy_service::service::Service::new(
-        psu_registration.as_slice(),
+        registration,
         service_context,
         power_policy_service::service::config::Config::default(),
     )));
 
     spawner.must_spawn(power_policy_task(
-        EventReceivers::new(
+        ArrayEventReceivers::new(
             [device0, device1],
             [
                 device0_event_channel.dyn_receiver(),
@@ -275,13 +283,13 @@ async fn run(spawner: Spawner) {
 
 #[embassy_executor::task]
 async fn power_policy_task(
-    psu_events: EventReceivers<
+    psu_events: ArrayEventReceivers<
         'static,
         2,
         DeviceType,
         channel::DynamicReceiver<'static, power_policy_interface::psu::event::EventData>,
     >,
-    power_policy: &'static Mutex<GlobalRawMutex, power_policy_service::service::Service<'static, DeviceType>>,
+    power_policy: &'static ServiceType,
 ) {
     power_policy_service::service::task::task(psu_events, power_policy).await;
 }

examples/std/src/bin/type_c/service.rs

@@ -5,13 +5,15 @@ use embassy_sync::mutex::Mutex;
 use embassy_sync::once_lock::OnceLock;
 use embassy_sync::pubsub::PubSubChannel;
 use embassy_time::Timer;
+use embedded_services::event::NoopSender;
 use embedded_services::{GlobalRawMutex, IntrusiveList};
 use embedded_usb_pd::GlobalPortId;
 use embedded_usb_pd::ado::Ado;
 use embedded_usb_pd::type_c::Current;
 use log::*;
 use power_policy_interface::psu;
-use power_policy_service::psu::EventReceivers;
+use power_policy_service::psu::ArrayEventReceivers;
+use power_policy_service::service::registration::ArrayRegistration;
 use static_cell::StaticCell;
 use std_examples::type_c::mock_controller;
 use std_examples::type_c::mock_controller::Wrapper;
@@ -30,6 +32,11 @@ const DELAY_MS: u64 = 1000;
 
 type DeviceType = Mutex<GlobalRawMutex, PowerProxyDevice<'static>>;
 
+type PowerPolicyServiceType = Mutex<
+    GlobalRawMutex,
+    power_policy_service::service::Service<'static, ArrayRegistration<'static, DeviceType, 1, NoopSender, 1>>,
+>;
+
 #[embassy_executor::task]
 async fn controller_task(
     wrapper: &'static Wrapper<'static>,
@@ -72,13 +79,14 @@ async fn task(spawner: Spawner) {
 
     let (wrapper, policy_receiver, controller, state) = create_wrapper(controller_context);
 
-    static POWER_POLICY_PSU_REGISTRATION: StaticCell<[&DeviceType; 1]> = StaticCell::new();
-    let psu_registration = POWER_POLICY_PSU_REGISTRATION.init([&wrapper.ports[0].proxy]);
+    let power_policy_registration = ArrayRegistration {
+        psus: [&wrapper.ports[0].proxy],
+        service_senders: [NoopSender],
+    };
 
-    static POWER_SERVICE: StaticCell<Mutex<GlobalRawMutex, power_policy_service::service::Service<DeviceType>>> =
-        StaticCell::new();
+    static POWER_SERVICE: StaticCell<PowerPolicyServiceType> = StaticCell::new();
     let power_service = POWER_SERVICE.init(Mutex::new(power_policy_service::service::Service::new(
-        psu_registration,
+        power_policy_registration,
         power_service_context,
         power_policy_service::service::config::Config::default(),
     )));
@@ -111,7 +119,7 @@ async fn task(spawner: Spawner) {
     let cfu_client = CfuClient::new(&CFU_CLIENT).await;
 
     spawner.must_spawn(power_policy_task(
-        EventReceivers::new([&wrapper.ports[0].proxy], [policy_receiver]),
+        ArrayEventReceivers::new([&wrapper.ports[0].proxy], [policy_receiver]),
         power_service,
     ));
     spawner.must_spawn(type_c_service_task(type_c_service, [wrapper], cfu_client));
@@ -143,8 +151,8 @@ async fn task(spawner: Spawner) {
 
 #[embassy_executor::task]
 async fn power_policy_task(
-    psu_events: EventReceivers<'static, 1, DeviceType, DynamicReceiver<'static, psu::event::EventData>>,
-    power_policy: &'static Mutex<GlobalRawMutex, power_policy_service::service::Service<'static, DeviceType>>,
+    psu_events: ArrayEventReceivers<'static, 1, DeviceType, DynamicReceiver<'static, psu::event::EventData>>,
+    power_policy: &'static PowerPolicyServiceType,
 ) {
     power_policy_service::service::task::task(psu_events, power_policy).await;
 }