type-c-service/ucsi: Add support for async events

* Port events signal UCSI connector change events. The type-C
  service implementation broadcasts these events to allow for OPM
  notification. The service also tracks pending events, reports them in
  the returned CCI, and clears them following an ACK_CC_CI.
* Move type-C service away from comms messages
* Update examples

Author: RobertZ2011

Cargo.lock

@@ -11,3 +11,23 @@ pub struct DebugAccessoryMessage {
     /// Connected
     pub connected: bool,
 }
+
+/// UCSI connector change message
+#[derive(Copy, Clone, Debug)]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
+pub struct UcsiCiMessage {
+    /// Port
+    pub port: GlobalPortId,
+    /// Notify OPM
+    pub notify_opm: bool,
+}
+
+/// Top-level comms message
+#[derive(Copy, Clone, Debug)]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
+pub enum CommsMessage {
+    /// Debug accessory message
+    DebugAccessory(DebugAccessoryMessage),
+    /// UCSI CCI message
+    UcsiCci(UcsiCiMessage),
+}

embedded-service/src/type_c/comms.rs

@@ -17,8 +17,9 @@ use super::{ATTN_VDM_LEN, ControllerId, OTHER_VDM_LEN, external};
 use crate::ipc::deferred;
 use crate::power::policy;
 use crate::type_c::Cached;
+use crate::type_c::comms::CommsMessage;
 use crate::type_c::event::{PortEvent, PortPending};
-use crate::{GlobalRawMutex, IntrusiveNode, error, intrusive_list, trace};
+use crate::{GlobalRawMutex, IntrusiveNode, broadcaster::immediate as broadcaster, error, intrusive_list, trace};
 
 /// maximum number of data objects in a VDM
 pub const MAX_NUM_DATA_OBJECTS: usize = 7; // 7 VDOs of 4 bytes each
@@ -674,6 +675,8 @@ struct Context {
     port_events: Signal<GlobalRawMutex, PortPending>,
     /// Channel for receiving commands to the type-C service
     external_command: deferred::Channel<GlobalRawMutex, external::Command, external::Response<'static>>,
+    /// Event broadcaster
+    broadcaster: broadcaster::Immediate<CommsMessage>,
 }
 
 impl Context {
@@ -682,6 +685,7 @@ impl Context {
             controllers: intrusive_list::IntrusiveList::new(),
             port_events: Signal::new(),
             external_command: deferred::Channel::new(),
+            broadcaster: broadcaster::Immediate::default(),
         }
     }
 
@@ -741,6 +745,13 @@ pub(super) async fn get_num_ports() -> usize {
         .fold(0, |acc, controller| acc + controller.num_ports())
 }
 
+/// Register a message receiver for type-C messages
+pub async fn register_message_receiver(
+    receiver: &'static broadcaster::Receiver<'_, CommsMessage>,
+) -> intrusive_list::Result<()> {
+    CONTEXT.get().await.broadcaster.register_receiver(receiver)
+}
+
 /// Default command timeout
 /// set to high value since this is intended to prevent an unresponsive device from blocking the service implementation
 const DEFAULT_TIMEOUT: Duration = Duration::from_millis(5000);
@@ -1214,6 +1225,11 @@ impl ContextToken {
             _ => Err(PdError::InvalidResponse),
         }
     }
+
+    /// Broadcast a type-C message to all subscribers
+    pub async fn broadcast_message(&self, message: CommsMessage) {
+        CONTEXT.get().await.broadcaster.broadcast(message).await;
+    }
 }
 
 /// Execute an external port command

embedded-service/src/type_c/controller.rs

@@ -444,6 +444,12 @@ impl From<PortPending> for u32 {
     }
 }
 
+impl Default for PortPending {
+    fn default() -> Self {
+        Self::none()
+    }
+}
+
 impl FromIterator<usize> for PortPending {
     fn from_iter<T: IntoIterator<Item = usize>>(iter: T) -> Self {
         let mut flags = PortPending::none();

embedded-service/src/type_c/event.rs

@@ -29,13 +29,9 @@ embassy-imxrt = { git = "https://github.com/OpenDevicePartnership/embassy-imxrt"
     "mimxrt685s",
 ] }
 
-embassy-embedded-hal = { version = "0.5.0", features = [
-    "defmt",
-] }
+embassy-embedded-hal = { version = "0.5.0", features = ["defmt"] }
 
-embassy-sync = { version = "0.7.2", features = [
-    "defmt",
-] }
+embassy-sync = { version = "0.7.2", features = ["defmt"] }
 embassy-executor = { version = "0.9.1", features = [
     "arch-cortex-m",
     "executor-thread",

examples/rt633/Cargo.lock

@@ -11,9 +11,9 @@ use embassy_imxrt::{bind_interrupts, peripherals};
 use embassy_sync::mutex::Mutex;
 use embassy_time::{self as _, Delay};
 use embedded_cfu_protocol::protocol_definitions::{FwUpdateOffer, FwUpdateOfferResponse, FwVersion, HostToken};
+use embedded_services::GlobalRawMutex;
 use embedded_services::power::policy::DeviceId as PowerId;
 use embedded_services::type_c::{self, Cached, ControllerId};
-use embedded_services::{GlobalRawMutex, comms};
 use embedded_services::{error, info};
 use embedded_usb_pd::GlobalPortId;
 use static_cell::StaticCell;
@@ -48,78 +48,6 @@ type Wrapper<'a> = Tps6699xWrapper<'a, GlobalRawMutex, BusDevice<'a>, Validator>
 type Controller<'a> = tps6699x::controller::Controller<GlobalRawMutex, BusDevice<'a>>;
 type Interrupt<'a> = tps6699x::Interrupt<'a, GlobalRawMutex, BusDevice<'a>>;
 
-/// Battery mock that receives messages from power policy
-mod battery {
-    use defmt::{info, trace};
-    use embedded_services::comms;
-    use embedded_services::power::policy;
-
-    pub struct Device {
-        pub tp: comms::Endpoint,
-    }
-
-    impl Device {
-        pub fn new() -> Self {
-            Self {
-                tp: comms::Endpoint::uninit(comms::EndpointID::Internal(comms::Internal::Battery)),
-            }
-        }
-    }
-
-    impl comms::MailboxDelegate for Device {
-        fn receive(&self, message: &comms::Message) -> Result<(), comms::MailboxDelegateError> {
-            trace!("Got message");
-
-            let message = message
-                .data
-                .get::<policy::CommsMessage>()
-                .ok_or(comms::MailboxDelegateError::MessageNotFound)?;
-
-            match message.data {
-                policy::CommsData::ConsumerDisconnected(id) => {
-                    info!("Consumer disconnected: {}", id.0);
-                    Ok(())
-                }
-                policy::CommsData::ConsumerConnected(id, capability) => {
-                    info!("Consumer connected: {} {:?}", id.0, capability);
-                    Ok(())
-                }
-                _ => Ok(()),
-            }
-        }
-    }
-}
-
-/// Debug accesory listener mock
-mod debug {
-    use defmt::{info, trace};
-    use embedded_services::comms;
-    use embedded_services::type_c;
-
-    pub struct Device {
-        pub tp: comms::Endpoint,
-    }
-
-    impl Device {
-        pub fn new() -> Self {
-            Self {
-                tp: comms::Endpoint::uninit(comms::EndpointID::Internal(comms::Internal::Usbc)),
-            }
-        }
-    }
-
-    impl comms::MailboxDelegate for Device {
-        fn receive(&self, message: &comms::Message) -> Result<(), comms::MailboxDelegateError> {
-            trace!("Got message");
-            if let Some(message) = message.data.get::<type_c::comms::DebugAccessoryMessage>() {
-                info!("Debug accessory message: {:?}", message);
-            }
-
-            Ok(())
-        }
-    }
-}
-
 #[embassy_executor::task]
 async fn pd_controller_task(controller: &'static Wrapper<'static>) {
     loop {
@@ -200,17 +128,6 @@ async fn main(spawner: Spawner) {
     pd_controller.register().await.unwrap();
     spawner.must_spawn(pd_controller_task(pd_controller));
 
-    static BATTERY: StaticCell<battery::Device> = StaticCell::new();
-    let battery = BATTERY.init(battery::Device::new());
-
-    comms::register_endpoint(battery, &battery.tp).await.unwrap();
-
-    static DEBUG_ACCESSORY: StaticCell<debug::Device> = StaticCell::new();
-    let debug_accessory = DEBUG_ACCESSORY.init(debug::Device::new());
-    comms::register_endpoint(debug_accessory, &debug_accessory.tp)
-        .await
-        .unwrap();
-
     // Sync our internal state with the hardware
     type_c::external::sync_controller_state(CONTROLLER0_ID).await.unwrap();