diff --git a/examples/rt685s-evk/src/bin/type_c.rs b/examples/rt685s-evk/src/bin/type_c.rs
index b4103711..bc04122a 100644
--- a/examples/rt685s-evk/src/bin/type_c.rs
+++ b/examples/rt685s-evk/src/bin/type_c.rs
@@ -47,7 +47,9 @@ type PortType = Mutex<
         'static,
         Tps6699xMutex<'static>,
         SharedStateType,
-        DynamicSender<'static, type_c_interface::service::event::PortEventData>,
+        type_c_interface::port::event::NonBlockingSenderNotifier<
+            DynamicSender<'static, type_c_interface::service::event::PortEventData>,
+        >,
         PowerNotifier<'static>,
         DynamicSender<'static, type_c_service::controller::event::Loopback>,
     >,
@@ -91,9 +93,15 @@ type TypeCServiceEventReceiverType = type_c_service::service::event_receiver::Ar
     PowerPolicyReceiverType,
 >;
 
-type TypeCServiceSenderType = NoopSender;
-type TypeCRegistrationType =
-    type_c_service::service::registration::ArrayRegistration<'static, PortType, PORT_COUNT, TypeCServiceSenderType, 1>;
+type TypeCServiceNotifierType =
+    type_c_interface::service::event::NonBlockingSenderNotifier<'static, PortType, NoopSender>;
+type TypeCRegistrationType = type_c_service::service::registration::ArrayRegistration<
+    'static,
+    PortType,
+    PORT_COUNT,
+    TypeCServiceNotifierType,
+    1,
+>;
 type TypeCServiceType = type_c_service::service::Service<'static, TypeCRegistrationType>;
 type PortEventReceiverType = PortEventReceiver<
     'static,
@@ -245,7 +253,7 @@ async fn main(spawner: Spawner) {
         Default::default(),
         TypeCRegistrationType {
             ports: [port0, port1],
-            service_senders: [NoopSender],
+            service_notifiers: [NoopSender.into()],
             port_data: [
                 PortData {
                     local_port: Some(LocalPortId(0)),
diff --git a/examples/rt685s-evk/src/bin/type_c_cfu.rs b/examples/rt685s-evk/src/bin/type_c_cfu.rs
index d750c4b7..5be1a476 100644
--- a/examples/rt685s-evk/src/bin/type_c_cfu.rs
+++ b/examples/rt685s-evk/src/bin/type_c_cfu.rs
@@ -63,7 +63,9 @@ type PortType = Mutex<
         'static,
         Tps6699xMutex<'static>,
         PortSharedStateType,
-        DynamicSender<'static, type_c_interface::service::event::PortEventData>,
+        type_c_interface::port::event::NonBlockingSenderNotifier<
+            DynamicSender<'static, type_c_interface::service::event::PortEventData>,
+        >,
         PowerNotifier<'static>,
         DynamicSender<'static, type_c_service::controller::event::Loopback>,
     >,
@@ -107,9 +109,15 @@ type TypeCServiceEventReceiverType = type_c_service::service::event_receiver::Ar
     PowerPolicyReceiverType,
 >;
 
-type TypeCServiceSenderType = NoopSender;
-type TypeCRegistrationType =
-    type_c_service::service::registration::ArrayRegistration<'static, PortType, PORT_COUNT, TypeCServiceSenderType, 1>;
+type TypeCServiceNotifierType =
+    type_c_interface::service::event::NonBlockingSenderNotifier<'static, PortType, NoopSender>;
+type TypeCRegistrationType = type_c_service::service::registration::ArrayRegistration<
+    'static,
+    PortType,
+    PORT_COUNT,
+    TypeCServiceNotifierType,
+    1,
+>;
 type TypeCServiceType = type_c_service::service::Service<'static, TypeCRegistrationType>;
 type PortEventReceiverType = PortEventReceiver<
     'static,
@@ -391,7 +399,7 @@ async fn main(spawner: Spawner) {
                     local_port: Some(LocalPortId(1)),
                 },
             ],
-            service_senders: [NoopSender],
+            service_notifiers: [NoopSender.into()],
         },
     )));
 
diff --git a/examples/std/src/lib/type_c/mock_controller.rs b/examples/std/src/lib/type_c/mock_controller.rs
index 6f413f30..09fe3698 100644
--- a/examples/std/src/lib/type_c/mock_controller.rs
+++ b/examples/std/src/lib/type_c/mock_controller.rs
@@ -352,7 +352,9 @@ pub type Port<'a> = type_c_service::controller::Port<
     'a,
     Mutex<GlobalRawMutex, Controller<'a>>,
     Mutex<GlobalRawMutex, SharedState>,
-    channel::DynamicSender<'a, type_c_interface::service::event::PortEventData>,
+    type_c_interface::port::event::NonBlockingSenderNotifier<
+        channel::DynamicSender<'a, type_c_interface::service::event::PortEventData>,
+    >,
     PowerNotifier<'a>,
     channel::DynamicSender<'a, type_c_service::controller::event::Loopback>,
 >;
diff --git a/type-c-interface/src/port/event.rs b/type-c-interface/src/port/event.rs
index a5b6b0eb..e08db018 100644
--- a/type-c-interface/src/port/event.rs
+++ b/type-c-interface/src/port/event.rs
@@ -6,9 +6,17 @@
 //! Consequently [`PortNotificationEventBitfield`] implements iterator traits to allow for processing these events as a stream.
 use bitfield::bitfield;
 
+use core::future::ready;
+
+use embedded_services::event::{NonBlockingSender, Sender};
+use embedded_usb_pd::ado::Ado;
+
+use crate::control::dp::DpStatus;
+use crate::control::pd::PortStatus;
 use crate::control::vdm::AttnVdm;
 use crate::control::vdm::OtherVdm;
-
+use crate::port::notification::{Error as NotificationError, Notifier};
+use crate::service::event::{PortEventData, StatusChangedData};
 bitfield! {
     /// Raw bitfield of possible port status events
     #[derive(Copy, Clone, PartialEq, Eq, Default)]
@@ -404,6 +412,130 @@ impl From<PortNotificationEventBitfield> for PortEventBitfield {
     }
 }
 
+/// New-type that implements the [`Notifier`] trait for any [`NonBlockingSender<PortEventData>`].
+///
+/// This allows the user to choose blocking/non-blocking behavior when a type supports both.
+pub struct NonBlockingSenderNotifier<S: NonBlockingSender<PortEventData>>(pub S);
+
+impl<S: NonBlockingSender<PortEventData>> Notifier for NonBlockingSenderNotifier<S> {
+    fn notify_status_changed(
+        &mut self,
+        status_event: PortStatusEventBitfield,
+        previous_status: PortStatus,
+        current_status: PortStatus,
+    ) -> impl Future<Output = Result<(), NotificationError>> {
+        ready(
+            self.0
+                .try_send(PortEventData::StatusChanged(StatusChangedData {
+                    status_event,
+                    previous_status,
+                    current_status,
+                }))
+                .ok_or(NotificationError::WouldBlock),
+        )
+    }
+
+    fn notify_alert(&mut self, alert: Ado) -> impl Future<Output = Result<(), NotificationError>> {
+        ready(
+            self.0
+                .try_send(PortEventData::Alert(alert))
+                .ok_or(NotificationError::WouldBlock),
+        )
+    }
+
+    fn notify_vdm(&mut self, vdm: VdmData) -> impl Future<Output = Result<(), NotificationError>> {
+        ready(
+            self.0
+                .try_send(PortEventData::Vdm(vdm))
+                .ok_or(NotificationError::WouldBlock),
+        )
+    }
+
+    fn notify_discover_mode_completed(&mut self) -> impl Future<Output = Result<(), NotificationError>> {
+        ready(
+            self.0
+                .try_send(PortEventData::DiscoverModeCompleted)
+                .ok_or(NotificationError::WouldBlock),
+        )
+    }
+
+    fn notify_usb_mux_error_recovery(&mut self) -> impl Future<Output = Result<(), NotificationError>> {
+        ready(
+            self.0
+                .try_send(PortEventData::UsbMuxErrorRecovery)
+                .ok_or(NotificationError::WouldBlock),
+        )
+    }
+
+    fn notify_dp_status_update(&mut self, status: DpStatus) -> impl Future<Output = Result<(), NotificationError>> {
+        ready(
+            self.0
+                .try_send(PortEventData::DpStatusUpdate(status))
+                .ok_or(NotificationError::WouldBlock),
+        )
+    }
+}
+
+impl<S: NonBlockingSender<PortEventData>> From<S> for NonBlockingSenderNotifier<S> {
+    fn from(sender: S) -> Self {
+        Self(sender)
+    }
+}
+
+/// New-type that implements the [`Notifier`] trait for any [`Sender<PortEventData>`].
+///
+/// This allows the user to choose blocking/non-blocking behavior when a type supports both.
+pub struct SenderNotifier<S: Sender<PortEventData>>(pub S);
+
+impl<S: Sender<PortEventData>> Notifier for SenderNotifier<S> {
+    async fn notify_status_changed(
+        &mut self,
+        status_event: PortStatusEventBitfield,
+        previous_status: PortStatus,
+        current_status: PortStatus,
+    ) -> Result<(), NotificationError> {
+        self.0
+            .send(PortEventData::StatusChanged(StatusChangedData {
+                status_event,
+                previous_status,
+                current_status,
+            }))
+            .await;
+        Ok(())
+    }
+
+    async fn notify_alert(&mut self, alert: Ado) -> Result<(), NotificationError> {
+        self.0.send(PortEventData::Alert(alert)).await;
+        Ok(())
+    }
+
+    async fn notify_vdm(&mut self, vdm: VdmData) -> Result<(), NotificationError> {
+        self.0.send(PortEventData::Vdm(vdm)).await;
+        Ok(())
+    }
+
+    async fn notify_discover_mode_completed(&mut self) -> Result<(), NotificationError> {
+        self.0.send(PortEventData::DiscoverModeCompleted).await;
+        Ok(())
+    }
+
+    async fn notify_usb_mux_error_recovery(&mut self) -> Result<(), NotificationError> {
+        self.0.send(PortEventData::UsbMuxErrorRecovery).await;
+        Ok(())
+    }
+
+    async fn notify_dp_status_update(&mut self, status: DpStatus) -> Result<(), NotificationError> {
+        self.0.send(PortEventData::DpStatusUpdate(status)).await;
+        Ok(())
+    }
+}
+
+impl<S: Sender<PortEventData>> From<S> for SenderNotifier<S> {
+    fn from(sender: S) -> Self {
+        Self(sender)
+    }
+}
+
 #[cfg(test)]
 #[allow(clippy::unwrap_used)]
 mod tests {
diff --git a/type-c-interface/src/port/mod.rs b/type-c-interface/src/port/mod.rs
index 7920f6db..a9a20d51 100644
--- a/type-c-interface/src/port/mod.rs
+++ b/type-c-interface/src/port/mod.rs
@@ -2,6 +2,7 @@
 pub mod electrical_disconnect;
 pub mod event;
 pub mod max_sink_voltage;
+pub mod notification;
 pub mod pd;
 pub mod power;
 pub mod retimer;
diff --git a/type-c-interface/src/port/notification.rs b/type-c-interface/src/port/notification.rs
new file mode 100644
index 00000000..8a655b07
--- /dev/null
+++ b/type-c-interface/src/port/notification.rs
@@ -0,0 +1,80 @@
+//! Traits and types for port notifications.
+
+use embedded_services::sync::Lockable;
+use embedded_usb_pd::PdError;
+use embedded_usb_pd::ado::Ado;
+
+use crate::control::dp::DpStatus;
+use crate::control::pd::PortStatus;
+use crate::port::event::{PortStatusEventBitfield, VdmData};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
+#[non_exhaustive]
+pub enum Error {
+    /// The requested operation would block
+    WouldBlock,
+}
+
+/// Port notifier trait
+pub trait Notifier {
+    /// Notify that a port's status has changed
+    fn notify_status_changed(
+        &mut self,
+        status_event: PortStatusEventBitfield,
+        previous_status: PortStatus,
+        current_status: PortStatus,
+    ) -> impl Future<Output = Result<(), Error>>;
+    /// Notify that a PD alert was received
+    fn notify_alert(&mut self, alert: Ado) -> impl Future<Output = Result<(), Error>>;
+    /// Notify of a VDM event
+    fn notify_vdm(&mut self, vdm: VdmData) -> impl Future<Output = Result<(), Error>>;
+    /// Notify that discover mode has completed
+    fn notify_discover_mode_completed(&mut self) -> impl Future<Output = Result<(), Error>>;
+    /// Notify of a USB mux error recovery
+    fn notify_usb_mux_error_recovery(&mut self) -> impl Future<Output = Result<(), Error>>;
+    /// Notify of a DisplayPort status update
+    fn notify_dp_status_update(&mut self, status: DpStatus) -> impl Future<Output = Result<(), Error>>;
+}
+
+/// Port notification handler
+pub trait NotificationHandler<'port> {
+    type Port: Lockable<Inner: crate::port::pd::Pd> + 'port;
+
+    /// Handle a notification that a port's status has changed
+    fn process_notify_status_changed(
+        &mut self,
+        port: &'port Self::Port,
+        status_event: PortStatusEventBitfield,
+        previous_status: PortStatus,
+        current_status: PortStatus,
+    ) -> impl Future<Output = Result<(), PdError>>;
+    /// Handle a notification that a PD alert was received
+    fn process_notify_alert(
+        &mut self,
+        port: &'port Self::Port,
+        alert: Ado,
+    ) -> impl Future<Output = Result<(), PdError>>;
+    /// Handle a notification of a VDM event
+    fn process_notify_vdm(
+        &mut self,
+        port: &'port Self::Port,
+        vdm: VdmData,
+    ) -> impl Future<Output = Result<(), PdError>>;
+    /// Handle a notification that discover mode has completed
+    fn process_notify_discover_mode_completed(
+        &mut self,
+        port: &'port Self::Port,
+    ) -> impl Future<Output = Result<(), PdError>>;
+    /// Handle a notification of a USB mux error recovery
+    fn process_notify_usb_mux_error_recovery(
+        &mut self,
+        port: &'port Self::Port,
+    ) -> impl Future<Output = Result<(), PdError>>;
+    /// Handle a notification of a DisplayPort status update
+    fn process_notify_dp_status_update(
+        &mut self,
+        port: &'port Self::Port,
+        status: DpStatus,
+    ) -> impl Future<Output = Result<(), PdError>>;
+}
diff --git a/type-c-interface/src/service/event.rs b/type-c-interface/src/service/event.rs
index c446441d..b9be143e 100644
--- a/type-c-interface/src/service/event.rs
+++ b/type-c-interface/src/service/event.rs
@@ -1,5 +1,9 @@
 //! Comms service message definitions
 
+use core::future::ready;
+use core::marker::PhantomData;
+
+use embedded_services::event::{NonBlockingSender, Sender};
 use embedded_services::sync::Lockable;
 use embedded_usb_pd::{GlobalPortId, ado::Ado};
 
@@ -9,6 +13,7 @@ use crate::{
         event::{PortStatusEventBitfield, VdmData},
         pd::Pd,
     },
+    service::notification::{Error as NotificationError, Notifier},
 };
 
 /// Struct containing data for a [`PortEventData::StatusChanged`] event
@@ -93,3 +98,134 @@ impl<'port, Port: Lockable<Inner: Pd>> Clone for Event<'port, Port> {
         }
     }
 }
+
+/// New-type that implements the [`Notifier`] trait for any [`NonBlockingSender<Event>`].
+///
+/// This allows the user to choose blocking/non-blocking behavior when a type supports both.
+pub struct NonBlockingSenderNotifier<'port, Port: Lockable<Inner: Pd> + 'port, S: NonBlockingSender<Event<'port, Port>>>
+{
+    pub sender: S,
+    _phantom: PhantomData<&'port Port>,
+}
+
+impl<'port, Port: Lockable<Inner: Pd>, S: NonBlockingSender<Event<'port, Port>>>
+    NonBlockingSenderNotifier<'port, Port, S>
+{
+    /// Create a new [`NonBlockingSenderNotifier`]
+    pub fn new(sender: S) -> Self {
+        Self {
+            sender,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<'port, Port: Lockable<Inner: Pd>, S: NonBlockingSender<Event<'port, Port>>> Notifier<'port>
+    for NonBlockingSenderNotifier<'port, Port, S>
+{
+    type Port = Port;
+
+    fn notify_debug_accessory(
+        &mut self,
+        port: &'port Self::Port,
+        connected: bool,
+    ) -> impl Future<Output = Result<(), NotificationError>> {
+        ready(
+            self.sender
+                .try_send(Event {
+                    port,
+                    event: EventData::DebugAccessory(DebugAccessoryData { connected }),
+                })
+                .ok_or(NotificationError::WouldBlock),
+        )
+    }
+
+    fn notify_ucsi_change_indicator(
+        &mut self,
+        port: &'port Self::Port,
+        port_id: GlobalPortId,
+        notify_opm: bool,
+    ) -> impl Future<Output = Result<(), NotificationError>> {
+        ready(
+            self.sender
+                .try_send(Event {
+                    port,
+                    event: EventData::UsciChangeIndicator(UsciChangeIndicatorData {
+                        port: port_id,
+                        notify_opm,
+                    }),
+                })
+                .ok_or(NotificationError::WouldBlock),
+        )
+    }
+}
+
+impl<'port, Port: Lockable<Inner: Pd>, S: NonBlockingSender<Event<'port, Port>>> From<S>
+    for NonBlockingSenderNotifier<'port, Port, S>
+{
+    fn from(sender: S) -> Self {
+        Self::new(sender)
+    }
+}
+
+/// New-type that implements the [`Notifier`] trait for any [`Sender<Event>`].
+///
+/// This allows the user to choose blocking/non-blocking behavior when a type supports both.
+pub struct SenderNotifier<'port, Port: Lockable<Inner: Pd> + 'port, S: Sender<Event<'port, Port>>> {
+    pub sender: S,
+    _phantom: PhantomData<&'port Port>,
+}
+
+impl<'port, Port: Lockable<Inner: Pd>, S: Sender<Event<'port, Port>>> SenderNotifier<'port, Port, S> {
+    /// Create a new [`SenderNotifier`]
+    pub fn new(sender: S) -> Self {
+        Self {
+            sender,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<'port, Port: Lockable<Inner: Pd>, S: Sender<Event<'port, Port>>> Notifier<'port>
+    for SenderNotifier<'port, Port, S>
+{
+    type Port = Port;
+
+    async fn notify_debug_accessory(
+        &mut self,
+        port: &'port Self::Port,
+        connected: bool,
+    ) -> Result<(), NotificationError> {
+        self.sender
+            .send(Event {
+                port,
+                event: EventData::DebugAccessory(DebugAccessoryData { connected }),
+            })
+            .await;
+        Ok(())
+    }
+
+    async fn notify_ucsi_change_indicator(
+        &mut self,
+        port: &'port Self::Port,
+        port_id: GlobalPortId,
+        notify_opm: bool,
+    ) -> Result<(), NotificationError> {
+        self.sender
+            .send(Event {
+                port,
+                event: EventData::UsciChangeIndicator(UsciChangeIndicatorData {
+                    port: port_id,
+                    notify_opm,
+                }),
+            })
+            .await;
+        Ok(())
+    }
+}
+
+impl<'port, Port: Lockable<Inner: Pd>, S: Sender<Event<'port, Port>>> From<S> for SenderNotifier<'port, Port, S> {
+    fn from(sender: S) -> Self {
+        Self::new(sender)
+    }
+}
diff --git a/type-c-interface/src/service/mod.rs b/type-c-interface/src/service/mod.rs
index 53f11265..7016cd88 100644
--- a/type-c-interface/src/service/mod.rs
+++ b/type-c-interface/src/service/mod.rs
@@ -1 +1,2 @@
 pub mod event;
+pub mod notification;
diff --git a/type-c-interface/src/service/notification.rs b/type-c-interface/src/service/notification.rs
new file mode 100644
index 00000000..7fa17b5b
--- /dev/null
+++ b/type-c-interface/src/service/notification.rs
@@ -0,0 +1,33 @@
+//! Traits and types for service notifications.
+
+use embedded_services::sync::Lockable;
+use embedded_usb_pd::GlobalPortId;
+
+use crate::port::pd::Pd;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[cfg_attr(feature = "defmt", derive(defmt::Format))]
+#[non_exhaustive]
+pub enum Error {
+    /// Implementation would block
+    WouldBlock,
+}
+
+/// Service notifier trait
+pub trait Notifier<'port> {
+    type Port: Lockable<Inner: Pd> + 'port;
+
+    /// Notify that a debug accessory was connected or disconnected
+    fn notify_debug_accessory(
+        &mut self,
+        port: &'port Self::Port,
+        connected: bool,
+    ) -> impl Future<Output = Result<(), Error>>;
+    /// Notify of a UCSI connector change
+    fn notify_ucsi_change_indicator(
+        &mut self,
+        port: &'port Self::Port,
+        port_id: GlobalPortId,
+        notify_opm: bool,
+    ) -> impl Future<Output = Result<(), Error>>;
+}
diff --git a/type-c-service/src/controller/electrical_disconnect.rs b/type-c-service/src/controller/electrical_disconnect.rs
index 6cf3f6b1..6dad4b6d 100644
--- a/type-c-service/src/controller/electrical_disconnect.rs
+++ b/type-c-service/src/controller/electrical_disconnect.rs
@@ -12,11 +12,11 @@ impl<
     'device,
     C: Lockable<Inner: Pd + ElectricalDisconnect>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
 > type_c_interface::port::electrical_disconnect::ElectricalDisconnect
-    for Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+    for Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     async fn execute_electrical_disconnect(&mut self, reconnect_time_s: Option<NonZeroU8>) -> Result<(), PdError> {
         self.controller
diff --git a/type-c-service/src/controller/macros.rs b/type-c-service/src/controller/macros.rs
index 4793d803..9f35b87b 100644
--- a/type-c-service/src/controller/macros.rs
+++ b/type-c-service/src/controller/macros.rs
@@ -52,6 +52,9 @@ macro_rules! define_controller_port_static_cell_channel {
 
             /// Type alias for the type-c service event sender
             pub type InnerTypeCSenderType = ::embassy_sync::channel::DynamicSender<'static, ::type_c_interface::service::event::PortEventData>;
+            /// Type alias for the type-c service event notifier
+            pub type InnerTypeCNotifierType =
+                ::type_c_interface::port::event::NonBlockingSenderNotifier<InnerTypeCSenderType>;
             /// Type alias for the type-c service event receiver
             pub type InnerTypeCReceiverType = ::embassy_sync::channel::DynamicReceiver<'static, ::type_c_interface::service::event::PortEventData>;
 
@@ -81,8 +84,8 @@ macro_rules! define_controller_port_static_cell_channel {
                     $controller,
                     // Shared state type
                     InnerSharedStateType,
-                    // Type-C service event sender type
-                    InnerTypeCSenderType,
+                    // Type-C service event notifier type
+                    InnerTypeCNotifierType,
                     // Power policy event sender type
                     InnerPowerPolicyNotifierType,
                     // Loopback event sender type
@@ -170,7 +173,7 @@ macro_rules! define_controller_port_static_cell_channel {
                     port,
                     controller,
                     shared_state,
-                    type_c_sender,
+                    type_c_sender.into(),
                     power_policy_sender.into(),
                     loopback_sender,
                 )));
diff --git a/type-c-service/src/controller/max_sink_voltage.rs b/type-c-service/src/controller/max_sink_voltage.rs
index 9c0c7263..c3ba6bdc 100644
--- a/type-c-service/src/controller/max_sink_voltage.rs
+++ b/type-c-service/src/controller/max_sink_voltage.rs
@@ -11,11 +11,11 @@ impl<
     'device,
     C: Lockable<Inner: Pd + MaxSinkVoltage>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
 > type_c_interface::port::max_sink_voltage::MaxSinkVoltage
-    for Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+    for Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     async fn set_max_sink_voltage(&mut self, voltage_mv: Option<u16>) -> Result<(), PdError> {
         // A change in the maximum sink voltage can trigger a PD renegotiation. During that transition the
diff --git a/type-c-service/src/controller/mod.rs b/type-c-service/src/controller/mod.rs
index 371db86a..8afa153c 100644
--- a/type-c-service/src/controller/mod.rs
+++ b/type-c-service/src/controller/mod.rs
@@ -28,7 +28,7 @@ pub struct Port<
     'device,
     C: Lockable<Inner: Pd>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
 > {
@@ -42,8 +42,8 @@ pub struct Port<
     name: &'static str,
     /// Cached port status
     status: PortStatus,
-    /// Sender for type-c service events
-    type_c_sender: TypeCSender,
+    /// Notifier for type-c service events
+    port_notifier: PortNotifier,
     /// Notifier for power policy events
     power_policy_notifier: PowerNotifier,
     /// Configuration
@@ -58,10 +58,10 @@ impl<
     'device,
     C: Lockable<Inner: Pd>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
-> Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+> Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     /// Create new Port instance
     // TODO: refactor arguments into a registration struct
@@ -72,7 +72,7 @@ impl<
         port: LocalPortId,
         controller: &'device C,
         shared_state: &'device Shared,
-        type_c_sender: TypeCSender,
+        port_notifier: PortNotifier,
         power_policy_notifier: PowerNotifier,
         loopback_sender: LoopbackSender,
     ) -> Self {
@@ -86,7 +86,7 @@ impl<
             config,
             shared_state,
             loopback_sender,
-            type_c_sender,
+            port_notifier,
         }
     }
 
@@ -148,16 +148,24 @@ impl<
         )
         .await?;
 
-        let event = ServicePortEventData::StatusChanged(StatusChangedData {
+        let status_changed = StatusChangedData {
             status_event,
             previous_status: self.status,
             current_status: new_status,
-        });
+        };
         self.status = new_status;
-        if self.type_c_sender.try_send(event).is_none() {
-            error!("Failed to send port status type-C event");
+        if let Err(e) = self
+            .port_notifier
+            .notify_status_changed(
+                status_changed.status_event,
+                status_changed.previous_status,
+                status_changed.current_status,
+            )
+            .await
+        {
+            error!("Failed to send port status type-C event: {:#?}", e);
         }
-        Ok(event)
+        Ok(ServicePortEventData::StatusChanged(status_changed))
     }
 
     /// Handle a plug event
@@ -225,10 +233,10 @@ impl<
     'device,
     C: Lockable<Inner: Pd>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
-> Named for Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+> Named for Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     fn name(&self) -> &'static str {
         self.name
diff --git a/type-c-service/src/controller/pd.rs b/type-c-service/src/controller/pd.rs
index e4d24088..2cfaba0f 100644
--- a/type-c-service/src/controller/pd.rs
+++ b/type-c-service/src/controller/pd.rs
@@ -22,10 +22,10 @@ impl<
     'device,
     C: Lockable<Inner: Pd>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
-> Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+> Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     /// Process a VDM event by retrieving the relevant VDM data from the `controller` for the appropriate `port`.
     pub(super) async fn process_vdm_event(
@@ -48,8 +48,8 @@ impl<
         };
 
         let event = ServicePortEventData::Vdm(vdm_data);
-        if self.type_c_sender.try_send(event).is_none() {
-            error!("Failed to send VDM type-C event");
+        if let Err(e) = self.port_notifier.notify_vdm(vdm_data).await {
+            error!("Failed to send VDM type-C event: {:#?}", e);
         }
         Ok(Some(event))
     }
@@ -59,8 +59,8 @@ impl<
         debug!("({}): Processing DP status update event", self.name);
         let status = self.controller.lock().await.get_dp_status(self.port).await?;
         let event = ServicePortEventData::DpStatusUpdate(status);
-        if self.type_c_sender.try_send(event).is_none() {
-            error!("Failed to send DP status update type-C event");
+        if let Err(e) = self.port_notifier.notify_dp_status_update(status).await {
+            error!("Failed to send DP status update type-C event: {:#?}", e);
         }
         Ok(event)
     }
@@ -70,8 +70,8 @@ impl<
         debug!("({}): PD alert: {:#?}", self.name, ado);
         if let Some(ado) = ado {
             let event = ServicePortEventData::Alert(ado);
-            if self.type_c_sender.try_send(event).is_none() {
-                error!("Failed to send PD alert type-C event");
+            if let Err(e) = self.port_notifier.notify_alert(ado).await {
+                error!("Failed to send PD alert type-C event: {:#?}", e);
             }
             Ok(Some(event))
         } else {
@@ -85,10 +85,10 @@ impl<
     'device,
     C: Lockable<Inner: Pd>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
-> type_c_interface::port::pd::Pd for Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+> type_c_interface::port::pd::Pd for Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     async fn get_port_status(&mut self) -> Result<PortStatus, PdError> {
         self.controller.lock().await.get_port_status(self.port).await
@@ -175,10 +175,10 @@ impl<
     'device,
     C: Lockable<Inner: Pd + StateMachine>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
-> type_c_interface::port::pd::StateMachine for Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+> type_c_interface::port::pd::StateMachine for Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     async fn set_pd_state_machine_config(&mut self, config: PdStateMachineConfig) -> Result<(), PdError> {
         self.controller
diff --git a/type-c-service/src/controller/power.rs b/type-c-service/src/controller/power.rs
index d689a536..aeb9f804 100644
--- a/type-c-service/src/controller/power.rs
+++ b/type-c-service/src/controller/power.rs
@@ -20,10 +20,10 @@ impl<
     'device,
     C: Lockable<Inner: Pd>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
-> Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+> Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     /// Handle a new contract as consumer
     pub(super) async fn process_new_consumer_contract(&mut self, new_status: &PortStatus) -> Result<(), PdError> {
@@ -180,10 +180,10 @@ impl<
     'device,
     C: Lockable<Inner: Pd>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
-> Psu for Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+> Psu for Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     async fn disconnect(&mut self) -> Result<(), PsuError> {
         self.controller
@@ -236,11 +236,11 @@ impl<
     'device,
     C: Lockable<Inner: Pd + SystemPowerStateStatus>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
 > type_c_interface::port::power::SystemPowerStateStatus
-    for Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+    for Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     async fn set_system_power_state_status(
         &mut self,
diff --git a/type-c-service/src/controller/retimer.rs b/type-c-service/src/controller/retimer.rs
index bea692e6..74114fd4 100644
--- a/type-c-service/src/controller/retimer.rs
+++ b/type-c-service/src/controller/retimer.rs
@@ -11,10 +11,10 @@ impl<
     'device,
     C: Lockable<Inner: Pd + Retimer>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
-> type_c_interface::port::retimer::Retimer for Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+> type_c_interface::port::retimer::Retimer for Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     async fn get_rt_fw_update_status(&mut self) -> Result<RetimerFwUpdateState, PdError> {
         self.controller.lock().await.get_rt_fw_update_status(self.port).await
diff --git a/type-c-service/src/controller/type_c.rs b/type-c-service/src/controller/type_c.rs
index ae38c351..740bdb8d 100644
--- a/type-c-service/src/controller/type_c.rs
+++ b/type-c-service/src/controller/type_c.rs
@@ -11,11 +11,11 @@ impl<
     'device,
     C: Lockable<Inner: Pd + StateMachine>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
 > type_c_interface::port::type_c::StateMachine
-    for Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+    for Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     async fn set_type_c_state_machine_config(&mut self, state: TypeCStateMachineState) -> Result<(), PdError> {
         self.controller
diff --git a/type-c-service/src/controller/ucsi.rs b/type-c-service/src/controller/ucsi.rs
index 53bf18c9..4eba6960 100644
--- a/type-c-service/src/controller/ucsi.rs
+++ b/type-c-service/src/controller/ucsi.rs
@@ -10,10 +10,10 @@ impl<
     'device,
     C: Lockable<Inner: Pd + UcsiLpm>,
     Shared: Lockable<Inner = SharedState>,
-    TypeCSender: NonBlockingSender<type_c_interface::service::event::PortEventData>,
+    PortNotifier: type_c_interface::port::notification::Notifier,
     PowerNotifier: power_policy_interface::psu::notification::Notifier,
     LoopbackSender: NonBlockingSender<event::Loopback>,
-> type_c_interface::ucsi::Lpm for Port<'device, C, Shared, TypeCSender, PowerNotifier, LoopbackSender>
+> type_c_interface::ucsi::Lpm for Port<'device, C, Shared, PortNotifier, PowerNotifier, LoopbackSender>
 {
     async fn execute_lpm_command(&mut self, command: lpm::LocalCommand) -> Result<Option<lpm::ResponseData>, PdError> {
         self.controller.lock().await.execute_lpm_command(command).await
diff --git a/type-c-service/src/service/mod.rs b/type-c-service/src/service/mod.rs
index 75db4b48..1181a035 100644
--- a/type-c-service/src/service/mod.rs
+++ b/type-c-service/src/service/mod.rs
@@ -1,19 +1,22 @@
 use core::marker::PhantomData;
 use core::ptr;
 
-use embedded_services::event::NonBlockingSender as _;
 use embedded_services::named::Named as _;
 use embedded_services::sync::Lockable;
 use embedded_services::{debug, error, info, trace};
 use embedded_usb_pd::GlobalPortId;
 use embedded_usb_pd::PdError as Error;
+use embedded_usb_pd::ado::Ado;
 use power_policy_interface::service::event::EventData as PowerPolicyEventData;
+use type_c_interface::control::dp::DpStatus;
 use type_c_interface::control::pd::PortStatus;
+use type_c_interface::port::event::VdmData;
+use type_c_interface::port::notification::NotificationHandler;
 use type_c_interface::port::pd::Pd;
-use type_c_interface::service::event::{DebugAccessoryData, EventData, PortEvent, PortEventData};
+use type_c_interface::service::event::{PortEvent, PortEventData};
+use type_c_interface::service::notification::Notifier as _;
 
 use type_c_interface::port::event::PortStatusEventBitfield;
-use type_c_interface::service::event::Event as ServiceEvent;
 
 use crate::service::registration::Registration;
 
@@ -75,10 +78,18 @@ impl<'port, Reg: Registration<'port>> Service<'port, Reg> {
     }
 
     /// Send an event to all registered listeners
-    fn broadcast_event(&mut self, event: ServiceEvent<'port, Reg::Port>) {
-        for sender in self.registration.event_senders() {
-            if sender.try_send(event.clone()).is_none() {
-                error!("Failed to send event to listener");
+    async fn notify_debug_accessory(&mut self, port: &'port Reg::Port, connected: bool) {
+        for notifier in self.registration.notifiers() {
+            if let Err(e) = notifier.notify_debug_accessory(port, connected).await {
+                error!("Failed to notify debug accessory: {:#?}", e);
+            }
+        }
+    }
+
+    async fn notify_ucsi_change_indicator(&mut self, port: &'port Reg::Port, port_id: GlobalPortId, notify_opm: bool) {
+        for notifier in self.registration.notifiers() {
+            if let Err(e) = notifier.notify_ucsi_change_indicator(port, port_id, notify_opm).await {
+                error!("Failed to notify UCSI change indicator: {:#?}", e);
             }
         }
     }
@@ -106,12 +117,7 @@ impl<'port, Reg: Registration<'port>> Service<'port, Reg> {
                 debug!("({}): Debug accessory disconnected", port_name);
             }
 
-            self.broadcast_event(ServiceEvent {
-                port,
-                event: EventData::DebugAccessory(DebugAccessoryData {
-                    connected: new_status.is_connected(),
-                }),
-            });
+            self.notify_debug_accessory(port, new_status.is_connected()).await;
         }
 
         self.handle_ucsi_port_event(port, GlobalPortId(self.get_port_index(port)? as u8), event, &new_status)
@@ -121,25 +127,22 @@ impl<'port, Reg: Registration<'port>> Service<'port, Reg> {
     }
 
     async fn process_port_event(&mut self, event: &PortEvent<'port, Reg::Port>) -> Result<(), Error> {
-        match &event.event {
-            PortEventData::StatusChanged(status_event) => {
-                self.process_port_status_event(
+        match event.event {
+            PortEventData::StatusChanged(data) => {
+                self.process_notify_status_changed(
                     event.port,
-                    status_event.status_event,
-                    status_event.current_status,
-                    status_event.previous_status,
+                    data.status_event,
+                    data.previous_status,
+                    data.current_status,
                 )
                 .await
             }
-            unhandled => {
-                // Currently just log notifications, but may want to do more in the future
-                debug!(
-                    "({}): Received notification event: {:#?}",
-                    event.port.lock().await.name(),
-                    unhandled
-                );
-                Ok(())
-            }
+            PortEventData::Alert(ado) => self.process_notify_alert(event.port, ado).await,
+            PortEventData::Vdm(vdm) => self.process_notify_vdm(event.port, vdm).await,
+            PortEventData::DiscoverModeCompleted => self.process_notify_discover_mode_completed(event.port).await,
+            PortEventData::UsbMuxErrorRecovery => self.process_notify_usb_mux_error_recovery(event.port).await,
+            PortEventData::DpStatusUpdate(status) => self.process_notify_dp_status_update(event.port, status).await,
+            _ => Ok(()),
         }
     }
 
@@ -157,3 +160,44 @@ impl<'port, Reg: Registration<'port>> Service<'port, Reg> {
         }
     }
 }
+
+impl<'port, Reg: Registration<'port>> NotificationHandler<'port> for Service<'port, Reg> {
+    type Port = Reg::Port;
+
+    async fn process_notify_status_changed(
+        &mut self,
+        port: &'port Reg::Port,
+        status_event: PortStatusEventBitfield,
+        previous_status: PortStatus,
+        current_status: PortStatus,
+    ) -> Result<(), Error> {
+        self.process_port_status_event(port, status_event, current_status, previous_status)
+            .await
+    }
+
+    async fn process_notify_alert(&mut self, port: &'port Reg::Port, alert: Ado) -> Result<(), Error> {
+        // Currently just log notifications, but may want to do more in the future
+        debug!("({}): Received PD alert: {:#?}", port.lock().await.name(), alert);
+        Ok(())
+    }
+
+    async fn process_notify_vdm(&mut self, port: &'port Reg::Port, vdm: VdmData) -> Result<(), Error> {
+        debug!("({}): Received VDM: {:#?}", port.lock().await.name(), vdm);
+        Ok(())
+    }
+
+    async fn process_notify_discover_mode_completed(&mut self, port: &'port Reg::Port) -> Result<(), Error> {
+        debug!("({}): Discover mode completed", port.lock().await.name());
+        Ok(())
+    }
+
+    async fn process_notify_usb_mux_error_recovery(&mut self, port: &'port Reg::Port) -> Result<(), Error> {
+        debug!("({}): USB mux error recovery", port.lock().await.name());
+        Ok(())
+    }
+
+    async fn process_notify_dp_status_update(&mut self, port: &'port Reg::Port, status: DpStatus) -> Result<(), Error> {
+        debug!("({}): DP status update: {:#?}", port.lock().await.name(), status);
+        Ok(())
+    }
+}
diff --git a/type-c-service/src/service/registration.rs b/type-c-service/src/service/registration.rs
index 654736f5..b7510d70 100644
--- a/type-c-service/src/service/registration.rs
+++ b/type-c-service/src/service/registration.rs
@@ -1,20 +1,20 @@
 //! Code related to registration with the type-C service
 
-use embedded_services::{event::NonBlockingSender, sync::Lockable};
+use embedded_services::sync::Lockable;
 use embedded_usb_pd::{GlobalPortId, LocalPortId};
 use type_c_interface::port::pd::Pd;
-use type_c_interface::service::event::Event as ServiceEvent;
+use type_c_interface::service::notification::Notifier as ServiceNotifierTrait;
 use type_c_interface::ucsi::Lpm as UcsiLpm;
 
 /// Registration trait that abstracts over various registration details.
 pub trait Registration<'port> {
     type Port: Lockable<Inner: Pd + UcsiLpm> + 'port;
-    type ServiceSender: NonBlockingSender<ServiceEvent<'port, Self::Port>>;
+    type ServiceNotifier: ServiceNotifierTrait<'port, Port = Self::Port>;
 
     /// Returns a slice to access ports
     fn ports(&self) -> &[&'port Self::Port];
-    /// Returns a slice to access type-c event senders
-    fn event_senders(&mut self) -> &mut [Self::ServiceSender];
+    /// Returns a slice to access type-c service notifiers
+    fn notifiers(&mut self) -> &mut [Self::ServiceNotifier];
     /// Returns the ucsi local port ID for a given global port
     fn ucsi_local_port_id(&self, global_port: GlobalPortId) -> Option<LocalPortId>;
 }
@@ -29,30 +29,30 @@ pub struct ArrayRegistration<
     'port,
     Port: Lockable<Inner: Pd + UcsiLpm> + 'port,
     const PORT_COUNT: usize,
-    ServiceSender: NonBlockingSender<ServiceEvent<'port, Port>>,
-    const SERVICE_SENDER_COUNT: usize,
+    ServiceNotifier: ServiceNotifierTrait<'port, Port = Port>,
+    const SERVICE_NOTIFIER_COUNT: usize,
 > {
     /// Array of registered ports
     pub ports: [&'port Port; PORT_COUNT],
     /// Array of local port data
     pub port_data: [PortData; PORT_COUNT],
-    /// Array of service event senders
-    pub service_senders: [ServiceSender; SERVICE_SENDER_COUNT],
+    /// Array of service event notifiers
+    pub service_notifiers: [ServiceNotifier; SERVICE_NOTIFIER_COUNT],
 }
 
 impl<
     'port,
     Port: Lockable<Inner: Pd + UcsiLpm> + 'port,
     const PORT_COUNT: usize,
-    ServiceSender: NonBlockingSender<ServiceEvent<'port, Port>>,
-    const SERVICE_SENDER_COUNT: usize,
-> Registration<'port> for ArrayRegistration<'port, Port, PORT_COUNT, ServiceSender, SERVICE_SENDER_COUNT>
+    ServiceNotifier: ServiceNotifierTrait<'port, Port = Port>,
+    const SERVICE_NOTIFIER_COUNT: usize,
+> Registration<'port> for ArrayRegistration<'port, Port, PORT_COUNT, ServiceNotifier, SERVICE_NOTIFIER_COUNT>
 {
     type Port = Port;
-    type ServiceSender = ServiceSender;
+    type ServiceNotifier = ServiceNotifier;
 
-    fn event_senders(&mut self) -> &mut [Self::ServiceSender] {
-        &mut self.service_senders
+    fn notifiers(&mut self) -> &mut [Self::ServiceNotifier] {
+        &mut self.service_notifiers
     }
 
     fn ports(&self) -> &[&'port Self::Port] {
diff --git a/type-c-service/src/service/ucsi.rs b/type-c-service/src/service/ucsi.rs
index c7996632..6083aeb2 100644
--- a/type-c-service/src/service/ucsi.rs
+++ b/type-c-service/src/service/ucsi.rs
@@ -8,7 +8,6 @@ use embedded_usb_pd::ucsi::ppm::state_machine::{
 };
 use embedded_usb_pd::ucsi::{GlobalCommand, ResponseData, lpm, ppm};
 use embedded_usb_pd::{PdError, PowerRole};
-use type_c_interface::service::event::{Event, UsciChangeIndicatorData};
 use type_c_interface::ucsi::Lpm as _;
 
 use super::*;
@@ -180,14 +179,8 @@ impl<'port, Reg: Registration<'port>> Service<'port, Reg> {
             return;
         };
 
-        self.broadcast_event(Event {
-            port,
-            event: EventData::UsciChangeIndicator(UsciChangeIndicatorData {
-                port: *next_port,
-                // False here because the OPM gets notified by the CCI, don't need a separate notification
-                notify_opm: false,
-            }),
-        });
+        // notify_opm is false here because the OPM gets notified by the CCI, no separate notification needed
+        self.notify_ucsi_change_indicator(port, *next_port, false).await;
 
         self.set_cci_connector_change(cci);
     }
@@ -381,13 +374,7 @@ impl<'port, Reg: Registration<'port>> Service<'port, Reg> {
         // of the CCI response to the ACK_CC_CI command. See [`Self::set_cci_connector_change`]
         let notify_opm = self.ucsi.pending_ports.is_empty();
         if self.ucsi.pending_ports.push_back(port_id).is_ok() {
-            self.broadcast_event(Event {
-                port,
-                event: EventData::UsciChangeIndicator(UsciChangeIndicatorData {
-                    port: port_id,
-                    notify_opm,
-                }),
-            });
+            self.notify_ucsi_change_indicator(port, port_id, notify_opm).await;
         } else {
             // This shouldn't happen because we have a single slot per port
             // Would likely indicate that an invalid port ID got in somehow
diff --git a/type-c-service/tests/common/mod.rs b/type-c-service/tests/common/mod.rs
index 291672f0..96dca1bb 100644
--- a/type-c-service/tests/common/mod.rs
+++ b/type-c-service/tests/common/mod.rs
@@ -35,6 +35,8 @@ pub type ControllerMockMutexType = Mutex<GlobalRawMutex, type_c_interface_test_m
 pub type PortTypeCSender<'a> = DynamicSender<'a, type_c_interface::service::event::PortEventData>;
 /// Corresponding receiver for [`PortTypeCSender`]
 pub type PortTypeCReceiver<'a> = DynamicReceiver<'a, type_c_interface::service::event::PortEventData>;
+/// Type-C port notification wrapper
+pub type PortTypeCNotifier<'a> = type_c_interface::port::event::NonBlockingSenderNotifier<PortTypeCSender<'a>>;
 /// [`type_c_service::controller::Port`] sender to power policy service
 pub type PortPowerSender<'a> = DynamicSender<'a, power_policy_interface::psu::event::EventData>;
 /// Corresponding receiver for [`PortPowerSender`]
@@ -61,7 +63,7 @@ pub type PortMutexType<'port, 'ch> = Mutex<
         // Shared state between the event receiver and port logic
         PortSharedState,
         // Sender to the type-C service
-        PortTypeCSender<'ch>,
+        PortTypeCNotifier<'ch>,
         // Power policy notifier
         PortPowerNotifier<'ch>,
         // Loopback sender
@@ -114,6 +116,12 @@ pub type TypeCServiceSender<'port, 'ch> =
 /// Receiver for events broadcast by the type-C service
 pub type TypeCServiceReceiver<'port, 'ch> =
     DynamicReceiver<'ch, type_c_interface::service::event::Event<'port, PortMutexType<'port, 'ch>>>;
+/// Notifier for events broadcast by the type-C service
+pub type TypeCServiceNotifier<'port, 'ch> = type_c_interface::service::event::NonBlockingSenderNotifier<
+    'port,
+    PortMutexType<'port, 'ch>,
+    TypeCServiceSender<'port, 'ch>,
+>;
 /// Type-C service registration type
 pub type TypeCRegistrationType<'port, 'ch> = type_c_service::service::registration::ArrayRegistration<
     'port,
@@ -121,9 +129,9 @@ pub type TypeCRegistrationType<'port, 'ch> = type_c_service::service::registrati
     PortMutexType<'port, 'ch>,
     // Number of type-C ports
     TYPE_C_PORT_COUNT,
-    // Senders for events broadcast by the service
-    TypeCServiceSender<'port, 'ch>,
-    // Number of registered service event senders
+    // Notifiers for events broadcast by the service
+    TypeCServiceNotifier<'port, 'ch>,
+    // Number of registered service event notifiers
     TYPE_C_SERVICE_SENDER_COUNT,
 >;
 /// Type-C service type
@@ -222,7 +230,7 @@ macro_rules! define_port {
                     $local_id,
                     &paste! { [<$name _mock>] },
                     &paste! { [<$name _shared_state>] },
-                    paste! { [<$name _type_c_sender>] },
+                    paste! { [<$name _type_c_sender>] }.into(),
                     paste! { [<$name _power_policy_sender>].into() },
                     paste! { [<$name _loopback_sender>] },
             )),
@@ -361,7 +369,7 @@ pub async fn run_test(
                     local_port: Some(LocalPortId(0)),
                 },
             ],
-            service_senders: [type_c_service_sender],
+            service_notifiers: [type_c_service_sender.into()],
         },
     ));