//! Event management and forwarding //! //! This module provides components to perform event routing. The most important component for this //! task is the [EventManager]. It receives all events and then routes them to event subscribers //! where appropriate. One common use case for satellite systems is to offer a light-weight //! publish-subscribe mechanism and IPC mechanism for software and hardware events which are also //! packaged as telemetry (TM) or can trigger a system response. //! //! It is recommended to read the //! [sat-rs book chapter](https://absatsw.irs.uni-stuttgart.de/projects/sat-rs/book/events.html) //! about events first: //! //! The event manager has a listener table abstracted by the [ListenerMapProvider], which maps //! listener groups identified by [ListenerKey]s to a [listener ID][ComponentId]. //! It also contains a sender table abstracted by the [SenderMapProvider] which maps these sender //! IDs to concrete [EventSendProvider]s. A simple approach would be to use one send event provider //! for each OBSW thread and then subscribe for all interesting events for a particular thread //! using the send event provider ID. //! //! This can be done with the [EventManager] like this: //! //! 1. Provide a concrete [EventReceiveProvider] implementation. This abstraction allow to use different //! message queue backends. A straightforward implementation where dynamic memory allocation is //! not a big concern could use [std::sync::mpsc::channel] to do this and is provided in //! form of the [MpscEventReceiver]. //! 2. To set up event creators, create channel pairs using some message queue implementation. //! Each event creator gets a (cloned) sender component which allows it to send events to the //! manager. //! 3. The event manager receives the receiver component as part of a [EventReceiveProvider] //! implementation so all events are routed to the manager. //! 4. Create the [event sender map][SenderMapProvider]s which allow routing events to //! subscribers. You can now use the subscriber component IDs to subscribe //! for event groups, for example by using the [EventManager::subscribe_single] method. //! 5. Add the send provider as well using the [EventManager::add_sender] call so the event //! manager can route listener groups to a the send provider. //! //! Some components like a PUS Event Service or PUS Event Action Service might require all //! events to package them as telemetry or start actions where applicable. //! Other components might only be interested in certain events. For example, a thermal system //! handler might only be interested in temperature events generated by a thermal sensor component. //! //! # Examples //! //! You can check [integration test](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs/tests/pus_events.rs) //! for a concrete example using multi-threading where events are routed to //! different threads. use crate::events::{EventU16, EventU32, GenericEvent, LargestEventRaw, LargestGroupIdRaw}; use crate::params::Params; use crate::queue::GenericSendError; use core::fmt::Debug; use core::marker::PhantomData; use core::slice::Iter; use crate::ComponentId; #[cfg(feature = "alloc")] pub use alloc_mod::*; #[cfg(feature = "std")] pub use std_mod::*; #[derive(PartialEq, Eq, Hash, Copy, Clone, Debug)] pub enum ListenerKey { Single(LargestEventRaw), Group(LargestGroupIdRaw), All, } #[derive(Debug)] pub struct EventMessage { sender_id: ComponentId, event: Event, params: Option, } impl EventMessage { pub fn new_generic( sender_id: ComponentId, event: Event, params: Option<&ParamProvider>, ) -> Self { Self { sender_id, event, params: params.cloned(), } } pub fn sender_id(&self) -> ComponentId { self.sender_id } pub fn event(&self) -> Event { self.event } pub fn params(&self) -> Option<&ParamProvider> { self.params.as_ref() } pub fn new(sender_id: ComponentId, event: Event) -> Self { Self::new_generic(sender_id, event, None) } pub fn new_with_params(sender_id: ComponentId, event: Event, params: &ParamProvider) -> Self { Self::new_generic(sender_id, event, Some(params)) } } pub type EventMessageU32 = EventMessage; pub type EventMessageU16 = EventMessage; /// Generic abstraction pub trait EventSendProvider { type Error; fn target_id(&self) -> ComponentId; fn send(&self, message: EventMessage) -> Result<(), Self::Error>; } /// Generic abstraction for an event receiver. pub trait EventReceiveProvider { type Error; /// This function has to be provided by any event receiver. A call may or may not return /// an event and optional auxiliary data. fn try_recv_event(&self) -> Result>, Self::Error>; } pub trait ListenerMapProvider { #[cfg(feature = "alloc")] fn get_listeners(&self) -> alloc::vec::Vec; fn contains_listener(&self, key: &ListenerKey) -> bool; fn get_listener_ids(&self, key: &ListenerKey) -> Option>; fn add_listener(&mut self, key: ListenerKey, listener_id: ComponentId) -> bool; fn remove_duplicates(&mut self, key: &ListenerKey); } pub trait SenderMapProvider< EventSender: EventSendProvider, Event: GenericEvent = EventU32, ParamProvider: Debug = Params, > { fn contains_send_event_provider(&self, target_id: &ComponentId) -> bool; fn get_send_event_provider(&self, target_id: &ComponentId) -> Option<&EventSender>; fn add_send_event_provider(&mut self, send_provider: EventSender) -> bool; } /// Generic event manager implementation. /// /// # Generics /// /// * `EventReceiver`: [EventReceiveProvider] used to receive all events. /// * `SenderMap`: [SenderMapProvider] which maps channel IDs to send providers. /// * `ListenerMap`: [ListenerMapProvider] which maps listener keys to channel IDs. /// * `EventSender`: [EventSendProvider] contained within the sender map which sends the events. /// * `Ev`: The event type. This type must implement the [GenericEvent]. Currently only [EventU32] /// and [EventU16] are supported. /// * `Data`: Auxiliary data which is sent with the event to provide optional context information pub struct EventManager< EventReceiver: EventReceiveProvider, SenderMap: SenderMapProvider, ListenerMap: ListenerMapProvider, EventSender: EventSendProvider, Event: GenericEvent = EventU32, ParamProvider: Debug = Params, > { event_receiver: EventReceiver, sender_map: SenderMap, listener_map: ListenerMap, phantom: core::marker::PhantomData<(EventSender, Event, ParamProvider)>, } #[derive(Debug)] pub enum EventRoutingResult { /// No event was received Empty, /// An event was received and routed to listeners. Handled { num_recipients: u32, event_msg: EventMessage, }, } #[derive(Debug)] pub enum EventRoutingError { Send(GenericSendError), NoSendersForKey(ListenerKey), NoSenderForId(ComponentId), } impl< EventReceiver: EventReceiveProvider, SenderMap: SenderMapProvider, ListenerMap: ListenerMapProvider, EventSender: EventSendProvider, Event: GenericEvent + Copy, ParamProvider: Debug, > EventManager { pub fn remove_duplicates(&mut self, key: &ListenerKey) { self.listener_map.remove_duplicates(key) } /// Subscribe for a unique event. pub fn subscribe_single(&mut self, event: &Event, sender_id: ComponentId) { self.update_listeners(ListenerKey::Single(event.raw_as_largest_type()), sender_id); } /// Subscribe for an event group. pub fn subscribe_group(&mut self, group_id: LargestGroupIdRaw, sender_id: ComponentId) { self.update_listeners(ListenerKey::Group(group_id), sender_id); } /// Subscribe for all events received by the manager. /// /// For example, this can be useful for a handler component which sends every event as /// a telemetry packet. pub fn subscribe_all(&mut self, sender_id: ComponentId) { self.update_listeners(ListenerKey::All, sender_id); } } impl< EventReceiver: EventReceiveProvider, SenderMap: SenderMapProvider, ListenerMap: ListenerMapProvider, EventSenderMap: EventSendProvider, Event: GenericEvent + Copy, ParamProvider: Debug, > EventManager { pub fn new_with_custom_maps( event_receiver: EventReceiver, sender_map: SenderMap, listener_map: ListenerMap, ) -> Self { EventManager { listener_map, sender_map, event_receiver, phantom: PhantomData, } } /// Add a new sender component which can be used to send events to subscribers. pub fn add_sender(&mut self, send_provider: EventSenderMap) { if !self .sender_map .contains_send_event_provider(&send_provider.target_id()) { self.sender_map.add_send_event_provider(send_provider); } } /// Generic function to update the event subscribers. fn update_listeners(&mut self, key: ListenerKey, sender_id: ComponentId) { self.listener_map.add_listener(key, sender_id); } } impl< EventReceiver: EventReceiveProvider, SenderMap: SenderMapProvider, ListenerMap: ListenerMapProvider, EventSenderMap: EventSendProvider, Event: GenericEvent + Copy, ParamProvider: Clone + Debug, > EventManager { /// This function will use the cached event receiver and try to receive one event. /// If an event was received, it will try to route that event to all subscribed event listeners. /// If this works without any issues, the [EventRoutingResult] will contain context information /// about the routed event. /// /// If an error occurs during the routing, the error handler will be called. The error handler /// should take a reference to the event message as the first argument, and the routing error /// as the second argument. pub fn try_event_handling, EventRoutingError)>( &self, mut error_handler: E, ) -> EventRoutingResult { let mut num_recipients = 0; let mut send_handler = |key: &ListenerKey, event_msg: &EventMessage| { if self.listener_map.contains_listener(key) { if let Some(ids) = self.listener_map.get_listener_ids(key) { for id in ids { if let Some(sender) = self.sender_map.get_send_event_provider(id) { if let Err(e) = sender.send(EventMessage::new_generic( *id, event_msg.event, event_msg.params.as_ref(), )) { error_handler(event_msg, EventRoutingError::Send(e)); } else { num_recipients += 1; } } else { error_handler(event_msg, EventRoutingError::NoSenderForId(*id)); } } } else { error_handler(event_msg, EventRoutingError::NoSendersForKey(*key)); } } }; if let Ok(Some(event_msg)) = self.event_receiver.try_recv_event() { let single_key = ListenerKey::Single(event_msg.event.raw_as_largest_type()); send_handler(&single_key, &event_msg); let group_key = ListenerKey::Group(event_msg.event.group_id_as_largest_type()); send_handler(&group_key, &event_msg); send_handler(&ListenerKey::All, &event_msg); return EventRoutingResult::Handled { num_recipients, event_msg, }; } EventRoutingResult::Empty } } #[cfg(feature = "alloc")] pub mod alloc_mod { use alloc::vec::Vec; use hashbrown::HashMap; use super::*; /// Helper type which constrains the sender map and listener map generics to the [DefaultSenderMap] /// and the [DefaultListenerMap]. It uses regular mpsc channels as the message queue backend. pub type EventManagerWithMpsc = EventManager< MpscEventReceiver, DefaultSenderMap, EV, AUX>, DefaultListenerMap, EventSenderMpsc, >; /// Helper type which constrains the sender map and listener map generics to the [DefaultSenderMap] /// and the [DefaultListenerMap]. It uses /// [bounded mpsc senders](https://doc.rust-lang.org/std/sync/mpsc/struct.SyncSender.html) as the /// message queue backend. pub type EventManagerWithBoundedMpsc = EventManager< MpscEventReceiver, DefaultSenderMap, Event, ParamProvider>, DefaultListenerMap, EventSenderMpscBounded, >; impl< EventReceiver: EventReceiveProvider, EventSender: EventSendProvider, Event: GenericEvent + Copy, ParamProvider: 'static + Debug, > EventManager< EventReceiver, DefaultSenderMap, DefaultListenerMap, EventSender, Event, ParamProvider, > { /// Create an event manager where the sender table will be the [DefaultSenderMap] /// and the listener table will be the [DefaultListenerMap]. pub fn new(event_receiver: EventReceiver) -> Self { Self { listener_map: DefaultListenerMap::default(), sender_map: DefaultSenderMap::default(), event_receiver, phantom: PhantomData, } } } /// Default listener map. /// /// Simple implementation which uses a [HashMap] and a [Vec] internally. #[derive(Default)] pub struct DefaultListenerMap { listeners: HashMap>, } impl ListenerMapProvider for DefaultListenerMap { fn get_listeners(&self) -> Vec { let mut key_list = Vec::new(); for key in self.listeners.keys() { key_list.push(*key); } key_list } fn contains_listener(&self, key: &ListenerKey) -> bool { self.listeners.contains_key(key) } fn get_listener_ids(&self, key: &ListenerKey) -> Option> { self.listeners.get(key).map(|vec| vec.iter()) } fn add_listener(&mut self, key: ListenerKey, sender_id: ComponentId) -> bool { if let Some(existing_list) = self.listeners.get_mut(&key) { existing_list.push(sender_id); } else { let new_list = alloc::vec![sender_id]; self.listeners.insert(key, new_list); } true } fn remove_duplicates(&mut self, key: &ListenerKey) { if let Some(list) = self.listeners.get_mut(key) { list.sort_unstable(); list.dedup(); } } } /// Default sender map. /// /// Simple implementation which uses a [HashMap] internally. pub struct DefaultSenderMap< EventSender: EventSendProvider, Event: GenericEvent = EventU32, ParamProvider: Debug = Params, > { senders: HashMap, phantom: PhantomData<(Event, ParamProvider)>, } impl< EventSender: EventSendProvider, Event: GenericEvent, ParamProvider: Debug, > Default for DefaultSenderMap { fn default() -> Self { Self { senders: Default::default(), phantom: Default::default(), } } } impl< EventSender: EventSendProvider, Event: GenericEvent, ParamProvider: Debug, > SenderMapProvider for DefaultSenderMap { fn contains_send_event_provider(&self, id: &ComponentId) -> bool { self.senders.contains_key(id) } fn get_send_event_provider(&self, id: &ComponentId) -> Option<&EventSender> { self.senders .get(id) .filter(|sender| sender.target_id() == *id) } fn add_send_event_provider(&mut self, send_provider: EventSender) -> bool { let id = send_provider.target_id(); if self.senders.contains_key(&id) { return false; } self.senders.insert(id, send_provider).is_none() } } } #[cfg(feature = "std")] pub mod std_mod { use crate::queue::GenericReceiveError; use super::*; use std::sync::mpsc; pub struct MpscEventReceiver { receiver: mpsc::Receiver>, } impl MpscEventReceiver { pub fn new(receiver: mpsc::Receiver>) -> Self { Self { receiver } } } impl EventReceiveProvider for MpscEventReceiver { type Error = GenericReceiveError; fn try_recv_event(&self) -> Result>, Self::Error> { match self.receiver.try_recv() { Ok(msg) => Ok(Some(msg)), Err(e) => match e { mpsc::TryRecvError::Empty => Ok(None), mpsc::TryRecvError::Disconnected => { Err(GenericReceiveError::TxDisconnected(None)) } }, } } } pub type MpscEventU32Receiver = MpscEventReceiver; pub type MpscEventU16Receiver = MpscEventReceiver; /// Generic event sender which uses a regular [mpsc::Sender] as the messaging backend to /// send events. #[derive(Clone)] pub struct EventSenderMpsc { target_id: ComponentId, sender: mpsc::Sender>, } impl EventSenderMpsc { pub fn new(target_id: ComponentId, sender: mpsc::Sender>) -> Self { Self { target_id, sender } } } impl EventSendProvider for EventSenderMpsc { type Error = GenericSendError; fn target_id(&self) -> ComponentId { self.target_id } fn send(&self, event_msg: EventMessage) -> Result<(), GenericSendError> { self.sender .send(event_msg) .map_err(|_| GenericSendError::RxDisconnected) } } /// Generic event sender which uses the [mpsc::SyncSender] as the messaging backend to send /// events. This has the advantage that the channel is bounded and thus more deterministic. #[derive(Clone)] pub struct EventSenderMpscBounded { target_id: ComponentId, sender: mpsc::SyncSender>, capacity: usize, } impl EventSenderMpscBounded { pub fn new( target_id: ComponentId, sender: mpsc::SyncSender>, capacity: usize, ) -> Self { Self { target_id, sender, capacity, } } } impl EventSendProvider for EventSenderMpscBounded { type Error = GenericSendError; fn target_id(&self) -> ComponentId { self.target_id } fn send(&self, event_msg: EventMessage) -> Result<(), Self::Error> { if let Err(e) = self.sender.try_send(event_msg) { return match e { mpsc::TrySendError::Full(_) => { Err(GenericSendError::QueueFull(Some(self.capacity as u32))) } mpsc::TrySendError::Disconnected(_) => Err(GenericSendError::RxDisconnected), }; } Ok(()) } } pub type EventU32SenderMpsc = EventSenderMpsc; pub type EventU16SenderMpsc = EventSenderMpsc; pub type EventU32SenderMpscBounded = EventSenderMpscBounded; pub type EventU16SenderMpscBounded = EventSenderMpscBounded; } #[cfg(test)] mod tests { use super::*; use crate::event_man::EventManager; use crate::events::{EventU32, GenericEvent, Severity}; use crate::params::{ParamsHeapless, ParamsRaw}; use crate::pus::test_util::{TEST_COMPONENT_ID_0, TEST_COMPONENT_ID_1}; use std::format; use std::sync::mpsc::{self}; const TEST_EVENT: EventU32 = EventU32::const_new(Severity::INFO, 0, 5); fn check_next_event( expected: EventU32, receiver: &mpsc::Receiver, ) -> Option { if let Ok(event_msg) = receiver.try_recv() { assert_eq!(event_msg.event, expected); return event_msg.params; } None } fn check_handled_event( res: EventRoutingResult, expected: EventU32, expected_num_sent: u32, ) { assert!(matches!(res, EventRoutingResult::Handled { .. })); if let EventRoutingResult::Handled { num_recipients, event_msg, } = res { assert_eq!(event_msg.event, expected); assert_eq!(num_recipients, expected_num_sent); } } fn generic_event_man() -> (mpsc::Sender, EventManagerWithMpsc) { let (event_sender, manager_queue) = mpsc::channel(); let event_man_receiver = MpscEventReceiver::new(manager_queue); (event_sender, EventManager::new(event_man_receiver)) } #[test] fn test_basic() { let (event_sender, mut event_man) = generic_event_man(); let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap(); let event_grp_1_0 = EventU32::new(Severity::HIGH, 1, 0).unwrap(); let (single_event_sender, single_event_receiver) = mpsc::channel(); let single_event_listener = EventSenderMpsc::new(0, single_event_sender); event_man.subscribe_single(&event_grp_0, single_event_listener.target_id()); event_man.add_sender(single_event_listener); let (group_event_sender_0, group_event_receiver_0) = mpsc::channel(); let group_event_listener = EventU32SenderMpsc::new(1, group_event_sender_0); event_man.subscribe_group(event_grp_1_0.group_id(), group_event_listener.target_id()); event_man.add_sender(group_event_listener); let error_handler = |event_msg: &EventMessageU32, e: EventRoutingError| { panic!("routing error occurred for event {:?}: {:?}", event_msg, e); }; // Test event with one listener event_sender .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_grp_0)) .expect("Sending single error failed"); let res = event_man.try_event_handling(&error_handler); // assert!(res.is_ok()); check_handled_event(res, event_grp_0, 1); check_next_event(event_grp_0, &single_event_receiver); // Test event which is sent to all group listeners event_sender .send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_grp_1_0)) .expect("Sending group error failed"); let res = event_man.try_event_handling(&error_handler); check_handled_event(res, event_grp_1_0, 1); check_next_event(event_grp_1_0, &group_event_receiver_0); } #[test] fn test_with_basic_params() { let error_handler = |event_msg: &EventMessageU32, e: EventRoutingError| { panic!("routing error occurred for event {:?}: {:?}", event_msg, e); }; let (event_sender, mut event_man) = generic_event_man(); let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap(); let (single_event_sender, single_event_receiver) = mpsc::channel(); let single_event_listener = EventSenderMpsc::new(0, single_event_sender); event_man.subscribe_single(&event_grp_0, single_event_listener.target_id()); event_man.add_sender(single_event_listener); event_sender .send(EventMessage::new_with_params( TEST_COMPONENT_ID_0.id(), event_grp_0, &Params::Heapless((2_u32, 3_u32).into()), )) .expect("Sending group error failed"); let res = event_man.try_event_handling(&error_handler); check_handled_event(res, event_grp_0, 1); let aux = check_next_event(event_grp_0, &single_event_receiver); assert!(aux.is_some()); let aux = aux.unwrap(); if let Params::Heapless(ParamsHeapless::Raw(ParamsRaw::U32Pair(pair))) = aux { assert_eq!(pair.0, 2); assert_eq!(pair.1, 3); } else { panic!("{}", format!("Unexpected auxiliary value type {:?}", aux)); } } /// Test listening for multiple groups #[test] fn test_multi_group() { let error_handler = |event_msg: &EventMessageU32, e: EventRoutingError| { panic!("routing error occurred for event {:?}: {:?}", event_msg, e); }; let (event_sender, mut event_man) = generic_event_man(); let res = event_man.try_event_handling(error_handler); assert!(matches!(res, EventRoutingResult::Empty)); let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap(); let event_grp_1_0 = EventU32::new(Severity::HIGH, 1, 0).unwrap(); let (event_grp_0_sender, event_grp_0_receiver) = mpsc::channel(); let event_grp_0_and_1_listener = EventU32SenderMpsc::new(0, event_grp_0_sender); event_man.subscribe_group( event_grp_0.group_id(), event_grp_0_and_1_listener.target_id(), ); event_man.subscribe_group( event_grp_1_0.group_id(), event_grp_0_and_1_listener.target_id(), ); event_man.add_sender(event_grp_0_and_1_listener); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_grp_0)) .expect("Sending Event Group 0 failed"); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_grp_1_0)) .expect("Sendign Event Group 1 failed"); let res = event_man.try_event_handling(error_handler); check_handled_event(res, event_grp_0, 1); let res = event_man.try_event_handling(error_handler); check_handled_event(res, event_grp_1_0, 1); check_next_event(event_grp_0, &event_grp_0_receiver); check_next_event(event_grp_1_0, &event_grp_0_receiver); } /// Test listening to the same event from multiple listeners. Also test listening /// to both group and single events from one listener #[test] fn test_listening_to_same_event_and_multi_type() { let error_handler = |event_msg: &EventMessageU32, e: EventRoutingError| { panic!("routing error occurred for event {:?}: {:?}", event_msg, e); }; let (event_sender, mut event_man) = generic_event_man(); let event_0 = EventU32::new(Severity::INFO, 0, 5).unwrap(); let event_1 = EventU32::new(Severity::HIGH, 1, 0).unwrap(); let (event_0_tx_0, event_0_rx_0) = mpsc::channel(); let (event_0_tx_1, event_0_rx_1) = mpsc::channel(); let event_listener_0 = EventU32SenderMpsc::new(0, event_0_tx_0); let event_listener_1 = EventU32SenderMpsc::new(1, event_0_tx_1); let event_listener_0_sender_id = event_listener_0.target_id(); event_man.subscribe_single(&event_0, event_listener_0_sender_id); event_man.add_sender(event_listener_0); let event_listener_1_sender_id = event_listener_1.target_id(); event_man.subscribe_single(&event_0, event_listener_1_sender_id); event_man.add_sender(event_listener_1); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_0)) .expect("Triggering Event 0 failed"); let res = event_man.try_event_handling(error_handler); check_handled_event(res, event_0, 2); check_next_event(event_0, &event_0_rx_0); check_next_event(event_0, &event_0_rx_1); event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_0)) .expect("Triggering Event 0 failed"); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_1)) .expect("Triggering Event 1 failed"); // 3 Events messages will be sent now let res = event_man.try_event_handling(error_handler); check_handled_event(res, event_0, 2); let res = event_man.try_event_handling(error_handler); check_handled_event(res, event_1, 1); // Both the single event and the group event should arrive now check_next_event(event_0, &event_0_rx_0); check_next_event(event_1, &event_0_rx_0); // Do double insertion and then remove duplicates event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id); event_man.remove_duplicates(&ListenerKey::Group(event_1.group_id())); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_1)) .expect("Triggering Event 1 failed"); let res = event_man.try_event_handling(error_handler); check_handled_event(res, event_1, 1); } #[test] fn test_all_events_listener() { let error_handler = |event_msg: &EventMessageU32, e: EventRoutingError| { panic!("routing error occurred for event {:?}: {:?}", event_msg, e); }; let (event_sender, manager_queue) = mpsc::channel(); let event_man_receiver = MpscEventReceiver::new(manager_queue); let mut event_man = EventManagerWithMpsc::new(event_man_receiver); let event_0 = EventU32::new(Severity::INFO, 0, 5).unwrap(); let event_1 = EventU32::new(Severity::HIGH, 1, 0).unwrap(); let (event_0_tx_0, all_events_rx) = mpsc::channel(); let all_events_listener = EventU32SenderMpsc::new(0, event_0_tx_0); event_man.subscribe_all(all_events_listener.target_id()); event_man.add_sender(all_events_listener); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), event_0)) .expect("Triggering event 0 failed"); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_1.id(), event_1)) .expect("Triggering event 1 failed"); let res = event_man.try_event_handling(error_handler); check_handled_event(res, event_0, 1); let res = event_man.try_event_handling(error_handler); check_handled_event(res, event_1, 1); check_next_event(event_0, &all_events_rx); check_next_event(event_1, &all_events_rx); } #[test] fn test_bounded_event_sender_queue_full() { let (event_sender, _event_receiver) = mpsc::sync_channel(3); let event_sender = EventU32SenderMpscBounded::new(1, event_sender, 3); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT)) .expect("sending test event failed"); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT)) .expect("sending test event failed"); event_sender .send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT)) .expect("sending test event failed"); let error = event_sender.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT)); if let Err(e) = error { assert!(matches!(e, GenericSendError::QueueFull(Some(3)))); } else { panic!("unexpected error {error:?}"); } } #[test] fn test_bounded_event_sender_rx_dropped() { let (event_sender, event_receiver) = mpsc::sync_channel(3); let event_sender = EventU32SenderMpscBounded::new(1, event_sender, 3); drop(event_receiver); if let Err(e) = event_sender.send(EventMessage::new(TEST_COMPONENT_ID_0.id(), TEST_EVENT)) { assert!(matches!(e, GenericSendError::RxDisconnected)); } else { panic!("Expected error"); } } }