power-policy: Broadcast events (#411)

This provides a more general way to receive events than the comms
service.

Author: RobertZ2011

embedded-service/src/power/policy/mod.rs

@@ -129,6 +129,8 @@ pub enum CommsData {
     Unconstrained(bool),
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
 /// Message to send with the comms service
 pub struct CommsMessage {
     /// Message data

embedded-service/src/power/policy/policy.rs

@@ -2,7 +2,8 @@
 use core::sync::atomic::{AtomicBool, Ordering};
 
 use crate::GlobalRawMutex;
-use crate::power::policy::{ConsumerPowerCapability, ProviderPowerCapability};
+use crate::broadcaster::immediate as broadcaster;
+use crate::power::policy::{CommsMessage, ConsumerPowerCapability, ProviderPowerCapability};
 use embassy_sync::channel::Channel;
 use embassy_sync::once_lock::OnceLock;
 
@@ -81,6 +82,8 @@ struct Context {
     policy_response: Channel<GlobalRawMutex, InternalResponseData, POLICY_CHANNEL_SIZE>,
     /// Registered chargers
     chargers: intrusive_list::IntrusiveList,
+    /// Message broadcaster
+    broadcaster: broadcaster::Immediate<CommsMessage>,
 }
 
 impl Context {
@@ -90,6 +93,7 @@ impl Context {
             chargers: intrusive_list::IntrusiveList::new(),
             policy_request: Channel::new(),
             policy_response: Channel::new(),
+            broadcaster: broadcaster::Immediate::default(),
         }
     }
 }
@@ -188,6 +192,13 @@ pub async fn check_chargers_ready() -> ChargerResponse {
     Ok(Ack)
 }
 
+/// Register a message receiver for power policy messages
+pub async fn register_message_receiver(
+    receiver: &'static broadcaster::Receiver<'_, CommsMessage>,
+) -> intrusive_list::Result<()> {
+    CONTEXT.get().await.broadcaster.register_receiver(receiver)
+}
+
 /// Singleton struct to give access to the power policy context
 pub struct ContextToken(());
 
@@ -255,4 +266,9 @@ impl ContextToken {
     pub async fn policy_action(&self, id: DeviceId) -> Result<action::policy::AnyState<'_>, Error> {
         Ok(self.get_device(id).await?.policy_action().await)
     }
+
+    /// Broadcast a power policy message to all subscribers
+    pub async fn broadcast_message(&self, message: CommsMessage) {
+        CONTEXT.get().await.broadcaster.broadcast(message).await;
+    }
 }

examples/std/src/bin/power_policy.rs

@@ -1,7 +1,10 @@
 use embassy_executor::{Executor, Spawner};
-use embassy_sync::once_lock::OnceLock;
+use embassy_sync::{blocking_mutex::raw::NoopRawMutex, once_lock::OnceLock, pubsub::PubSubChannel};
 use embassy_time::{self as _, Timer};
-use embedded_services::power::policy::{self, ConsumerPowerCapability, PowerCapability, device, flags};
+use embedded_services::{
+    broadcaster::immediate as broadcaster,
+    power::policy::{self, ConsumerPowerCapability, PowerCapability, device, flags},
+};
 use log::*;
 use static_cell::StaticCell;
 
@@ -186,6 +189,31 @@ async fn run(spawner: Spawner) {
     Timer::after_millis(250).await;
 }
 
+#[embassy_executor::task]
+async fn receiver_task() {
+    static CHANNEL: StaticCell<PubSubChannel<NoopRawMutex, policy::CommsMessage, 1, 1, 0>> = StaticCell::new();
+    let channel = CHANNEL.init(PubSubChannel::new());
+
+    let publisher = channel.dyn_immediate_publisher();
+    let mut subscriber = channel.dyn_subscriber().unwrap();
+
+    static RECEIVER: StaticCell<broadcaster::Receiver<'static, policy::CommsMessage>> = StaticCell::new();
+    let receiver = RECEIVER.init(broadcaster::Receiver::new(publisher));
+
+    policy::policy::register_message_receiver(receiver).await.unwrap();
+
+    loop {
+        match subscriber.next_message().await {
+            embassy_sync::pubsub::WaitResult::Message(msg) => {
+                info!("Received message: {msg:?}");
+            }
+            embassy_sync::pubsub::WaitResult::Lagged(count) => {
+                warn!("Lagged messages: {count}");
+            }
+        }
+    }
+}
+
 fn main() {
     env_logger::builder().filter_level(log::LevelFilter::Trace).init();
 
@@ -196,5 +224,6 @@ fn main() {
             power_policy_service::config::Config::default(),
         ));
         spawner.must_spawn(run(spawner));
+        spawner.must_spawn(receiver_task());
     });
 }

power-policy-service/src/lib.rs

@@ -86,6 +86,7 @@ impl PowerPolicy {
 
     /// Send a notification with the comms service
     async fn comms_notify(&self, message: CommsMessage) {
+        self.context.broadcast_message(message).await;
         let _ = self
             .tp
             .send(comms::EndpointID::Internal(comms::Internal::Battery), &message)