satrs_core/
event_man.rs

1//! Event management and forwarding
2//!
3//! This module provides components to perform event routing. The most important component for this
4//! task is the [EventManager]. It receives all events and then routes them to event subscribers
5//! where appropriate. One common use case for satellite systems is to offer a light-weight
6//! publish-subscribe mechanism and IPC mechanism for software and hardware events which are also
7//! packaged as telemetry (TM) or can trigger a system response.
8//!
9//! It is recommended to read the
10//! [sat-rs book chapter](https://absatsw.irs.uni-stuttgart.de/projects/sat-rs/book/events.html)
11//! about events first:
12//!
13//! The event manager has a listener table abstracted by the [ListenerTable], which maps
14//! listener groups identified by [ListenerKey]s to a [sender ID][ChannelId].
15//! It also contains a sender table abstracted by the [SenderTable] which maps these sender IDs
16//! to a concrete [SendEventProvider]s. A simple approach would be to use one send event provider
17//! for each OBSW thread and then subscribe for all interesting events for a particular thread
18//! using the send event provider ID.
19//!
20//! This can be done with the [EventManager] like this:
21//!
22//!  1. Provide a concrete [EventReceiver] implementation. This abstraction allow to use different
23//!     message queue backends. A straightforward implementation where dynamic memory allocation is
24//!     not a big concern could use [std::sync::mpsc::channel] to do this and is provided in
25//!     form of the [MpscEventReceiver].
26//!  2. To set up event creators, create channel pairs using some message queue implementation.
27//!     Each event creator gets a (cloned) sender component which allows it to send events to the
28//!     manager.
29//!  3. The event manager receives the receiver component as part of a [EventReceiver]
30//!     implementation so all events are routed to the manager.
31//!  4. Create the [send event providers][SendEventProvider]s which allow routing events to
32//!     subscribers. You can now use their [sender IDs][SendEventProvider::id] to subscribe for
33//!     event groups, for example by using the [EventManager::subscribe_single] method.
34//!  5. Add the send provider as well using the [EventManager::add_sender] call so the event
35//!     manager can route listener groups to a the send provider.
36//!
37//! Some components like a PUS Event Service or PUS Event Action Service might require all
38//! events to package them as telemetry or start actions where applicable.
39//! Other components might only be interested in certain events. For example, a thermal system
40//! handler might only be interested in temperature events generated by a thermal sensor component.
41//!
42//! # Examples
43//!
44//! You can check [integration test](https://egit.irs.uni-stuttgart.de/rust/sat-rs/src/branch/main/satrs-core/tests/pus_events.rs)
45//! for a concrete example using multi-threading where events are routed to
46//! different threads.
47use crate::events::{EventU16, EventU32, GenericEvent, LargestEventRaw, LargestGroupIdRaw};
48use crate::params::{Params, ParamsHeapless};
49#[cfg(feature = "alloc")]
50use alloc::boxed::Box;
51#[cfg(feature = "alloc")]
52use alloc::vec;
53#[cfg(feature = "alloc")]
54use alloc::vec::Vec;
55use core::slice::Iter;
56#[cfg(feature = "alloc")]
57use hashbrown::HashMap;
58
59use crate::ChannelId;
60#[cfg(feature = "std")]
61pub use stdmod::*;
62
63#[derive(PartialEq, Eq, Hash, Copy, Clone, Debug)]
64pub enum ListenerKey {
65    Single(LargestEventRaw),
66    Group(LargestGroupIdRaw),
67    All,
68}
69
70pub type EventWithHeaplessAuxData<Event> = (Event, Option<ParamsHeapless>);
71pub type EventU32WithHeaplessAuxData = EventWithHeaplessAuxData<EventU32>;
72pub type EventU16WithHeaplessAuxData = EventWithHeaplessAuxData<EventU16>;
73
74pub type EventWithAuxData<Event> = (Event, Option<Params>);
75pub type EventU32WithAuxData = EventWithAuxData<EventU32>;
76pub type EventU16WithAuxData = EventWithAuxData<EventU16>;
77
78pub trait SendEventProvider<Provider: GenericEvent, AuxDataProvider = Params> {
79    type Error;
80
81    fn id(&self) -> ChannelId;
82    fn send_no_data(&self, event: Provider) -> Result<(), Self::Error> {
83        self.send(event, None)
84    }
85    fn send(&self, event: Provider, aux_data: Option<AuxDataProvider>) -> Result<(), Self::Error>;
86}
87
88/// Generic abstraction for an event receiver.
89pub trait EventReceiver<Event: GenericEvent, AuxDataProvider = Params> {
90    /// This function has to be provided by any event receiver. A receive call may or may not return
91    /// an event.
92    ///
93    /// To allow returning arbitrary additional auxiliary data, a mutable slice is passed to the
94    /// [Self::receive] call as well. Receivers can write data to this slice, but care must be taken
95    /// to avoid panics due to size missmatches or out of bound writes.
96    fn receive(&self) -> Option<(Event, Option<AuxDataProvider>)>;
97}
98
99pub trait ListenerTable {
100    fn get_listeners(&self) -> Vec<ListenerKey>;
101    fn contains_listener(&self, key: &ListenerKey) -> bool;
102    fn get_listener_ids(&self, key: &ListenerKey) -> Option<Iter<ChannelId>>;
103    fn add_listener(&mut self, key: ListenerKey, sender_id: ChannelId) -> bool;
104    fn remove_duplicates(&mut self, key: &ListenerKey);
105}
106
107pub trait SenderTable<SendProviderError, Event: GenericEvent = EventU32, AuxDataProvider = Params> {
108    fn contains_send_event_provider(&self, id: &ChannelId) -> bool;
109    fn get_send_event_provider(
110        &self,
111        id: &ChannelId,
112    ) -> Option<&dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>>;
113    fn add_send_event_provider(
114        &mut self,
115        send_provider: Box<
116            dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>,
117        >,
118    ) -> bool;
119}
120
121/// Generic event manager implementation.
122///
123/// # Generics
124///
125///  * `SendProviderError`: [SendEventProvider] error type
126///  * `Event`: Concrete event provider, currently either [EventU32] or [EventU16]
127///  * `AuxDataProvider`: Concrete auxiliary data provider, currently either [Params] or
128///     [ParamsHeapless]
129pub struct EventManager<SendProviderError, Event: GenericEvent = EventU32, AuxDataProvider = Params>
130{
131    listener_table: Box<dyn ListenerTable>,
132    sender_table: Box<dyn SenderTable<SendProviderError, Event, AuxDataProvider>>,
133    event_receiver: Box<dyn EventReceiver<Event, AuxDataProvider>>,
134}
135
136/// Safety: It is safe to implement [Send] because all fields in the [EventManager] are [Send]
137/// as well
138#[cfg(feature = "std")]
139unsafe impl<E, Event: GenericEvent + Send, AuxDataProvider: Send> Send
140    for EventManager<E, Event, AuxDataProvider>
141{
142}
143
144#[cfg(feature = "std")]
145pub type EventManagerWithMpscQueue<Event, AuxDataProvider> = EventManager<
146    std::sync::mpsc::SendError<(Event, Option<AuxDataProvider>)>,
147    Event,
148    AuxDataProvider,
149>;
150
151#[derive(Debug)]
152pub enum EventRoutingResult<Event: GenericEvent, AuxDataProvider> {
153    /// No event was received
154    Empty,
155    /// An event was received and routed.
156    /// The first tuple entry will contain the number of recipients.
157    Handled(u32, Event, Option<AuxDataProvider>),
158}
159
160#[derive(Debug)]
161pub enum EventRoutingError<E> {
162    SendError(E),
163    NoSendersForKey(ListenerKey),
164    NoSenderForId(ChannelId),
165}
166
167#[derive(Debug)]
168pub struct EventRoutingErrorsWithResult<Event: GenericEvent, AuxDataProvider, E> {
169    pub result: EventRoutingResult<Event, AuxDataProvider>,
170    pub errors: [Option<EventRoutingError<E>>; 3],
171}
172
173impl<E, Event: GenericEvent + Copy> EventManager<E, Event> {
174    pub fn remove_duplicates(&mut self, key: &ListenerKey) {
175        self.listener_table.remove_duplicates(key)
176    }
177
178    /// Subscribe for a unique event.
179    pub fn subscribe_single(&mut self, event: &Event, sender_id: ChannelId) {
180        self.update_listeners(ListenerKey::Single(event.raw_as_largest_type()), sender_id);
181    }
182
183    /// Subscribe for an event group.
184    pub fn subscribe_group(&mut self, group_id: LargestGroupIdRaw, sender_id: ChannelId) {
185        self.update_listeners(ListenerKey::Group(group_id), sender_id);
186    }
187
188    /// Subscribe for all events received by the manager.
189    ///
190    /// For example, this can be useful for a handler component which sends every event as
191    /// a telemetry packet.
192    pub fn subscribe_all(&mut self, sender_id: ChannelId) {
193        self.update_listeners(ListenerKey::All, sender_id);
194    }
195}
196
197impl<E: 'static, Event: GenericEvent + Copy + 'static, AuxDataProvider: Clone + 'static>
198    EventManager<E, Event, AuxDataProvider>
199{
200    /// Create an event manager where the sender table will be the [DefaultSenderTableProvider]
201    /// and the listener table will be the [DefaultListenerTableProvider].
202    pub fn new(event_receiver: Box<dyn EventReceiver<Event, AuxDataProvider>>) -> Self {
203        let listener_table: Box<DefaultListenerTableProvider> = Box::default();
204        let sender_table: Box<DefaultSenderTableProvider<E, Event, AuxDataProvider>> =
205            Box::default();
206        Self::new_custom_tables(listener_table, sender_table, event_receiver)
207    }
208}
209
210impl<E, Event: GenericEvent + Copy, AuxDataProvider: Clone>
211    EventManager<E, Event, AuxDataProvider>
212{
213    pub fn new_custom_tables(
214        listener_table: Box<dyn ListenerTable>,
215        sender_table: Box<dyn SenderTable<E, Event, AuxDataProvider>>,
216        event_receiver: Box<dyn EventReceiver<Event, AuxDataProvider>>,
217    ) -> Self {
218        EventManager {
219            listener_table,
220            sender_table,
221            event_receiver,
222        }
223    }
224
225    pub fn add_sender(
226        &mut self,
227        send_provider: impl SendEventProvider<Event, AuxDataProvider, Error = E> + 'static,
228    ) {
229        if !self
230            .sender_table
231            .contains_send_event_provider(&send_provider.id())
232        {
233            self.sender_table
234                .add_send_event_provider(Box::new(send_provider));
235        }
236    }
237
238    fn update_listeners(&mut self, key: ListenerKey, sender_id: ChannelId) {
239        self.listener_table.add_listener(key, sender_id);
240    }
241
242    /// This function will use the cached event receiver and try to receive one event.
243    /// If an event was received, it will try to route that event to all subscribed event listeners.
244    /// If this works without any issues, the [EventRoutingResult] will contain context information
245    /// about the routed event.
246    ///
247    /// This function will track up to 3 errors returned as part of the
248    /// [EventRoutingErrorsWithResult] error struct.
249    pub fn try_event_handling(
250        &self,
251    ) -> Result<
252        EventRoutingResult<Event, AuxDataProvider>,
253        EventRoutingErrorsWithResult<Event, AuxDataProvider, E>,
254    > {
255        let mut err_idx = 0;
256        let mut err_slice = [None, None, None];
257        let mut num_recipients = 0;
258        let mut add_error = |error: EventRoutingError<E>| {
259            if err_idx < 3 {
260                err_slice[err_idx] = Some(error);
261                err_idx += 1;
262            }
263        };
264        let mut send_handler =
265            |key: &ListenerKey, event: Event, aux_data: &Option<AuxDataProvider>| {
266                if self.listener_table.contains_listener(key) {
267                    if let Some(ids) = self.listener_table.get_listener_ids(key) {
268                        for id in ids {
269                            if let Some(sender) = self.sender_table.get_send_event_provider(id) {
270                                if let Err(e) = sender.send(event, aux_data.clone()) {
271                                    add_error(EventRoutingError::SendError(e));
272                                } else {
273                                    num_recipients += 1;
274                                }
275                            } else {
276                                add_error(EventRoutingError::NoSenderForId(*id));
277                            }
278                        }
279                    } else {
280                        add_error(EventRoutingError::NoSendersForKey(*key));
281                    }
282                }
283            };
284        if let Some((event, aux_data)) = self.event_receiver.receive() {
285            let single_key = ListenerKey::Single(event.raw_as_largest_type());
286            send_handler(&single_key, event, &aux_data);
287            let group_key = ListenerKey::Group(event.group_id_as_largest_type());
288            send_handler(&group_key, event, &aux_data);
289            send_handler(&ListenerKey::All, event, &aux_data);
290            if err_idx > 0 {
291                return Err(EventRoutingErrorsWithResult {
292                    result: EventRoutingResult::Handled(num_recipients, event, aux_data),
293                    errors: err_slice,
294                });
295            }
296            return Ok(EventRoutingResult::Handled(num_recipients, event, aux_data));
297        }
298        Ok(EventRoutingResult::Empty)
299    }
300}
301
302#[derive(Default)]
303pub struct DefaultListenerTableProvider {
304    listeners: HashMap<ListenerKey, Vec<ChannelId>>,
305}
306
307pub struct DefaultSenderTableProvider<
308    SendProviderError,
309    Event: GenericEvent = EventU32,
310    AuxDataProvider = Params,
311> {
312    senders: HashMap<
313        ChannelId,
314        Box<dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>>,
315    >,
316}
317
318impl<SendProviderError, Event: GenericEvent, AuxDataProvider> Default
319    for DefaultSenderTableProvider<SendProviderError, Event, AuxDataProvider>
320{
321    fn default() -> Self {
322        Self {
323            senders: HashMap::new(),
324        }
325    }
326}
327
328impl ListenerTable for DefaultListenerTableProvider {
329    fn get_listeners(&self) -> Vec<ListenerKey> {
330        let mut key_list = Vec::new();
331        for key in self.listeners.keys() {
332            key_list.push(*key);
333        }
334        key_list
335    }
336
337    fn contains_listener(&self, key: &ListenerKey) -> bool {
338        self.listeners.contains_key(key)
339    }
340
341    fn get_listener_ids(&self, key: &ListenerKey) -> Option<Iter<ChannelId>> {
342        self.listeners.get(key).map(|vec| vec.iter())
343    }
344
345    fn add_listener(&mut self, key: ListenerKey, sender_id: ChannelId) -> bool {
346        if let Some(existing_list) = self.listeners.get_mut(&key) {
347            existing_list.push(sender_id);
348        } else {
349            let new_list = vec![sender_id];
350            self.listeners.insert(key, new_list);
351        }
352        true
353    }
354
355    fn remove_duplicates(&mut self, key: &ListenerKey) {
356        if let Some(list) = self.listeners.get_mut(key) {
357            list.sort_unstable();
358            list.dedup();
359        }
360    }
361}
362
363impl<SendProviderError, Event: GenericEvent, AuxDataProvider>
364    SenderTable<SendProviderError, Event, AuxDataProvider>
365    for DefaultSenderTableProvider<SendProviderError, Event, AuxDataProvider>
366{
367    fn contains_send_event_provider(&self, id: &ChannelId) -> bool {
368        self.senders.contains_key(id)
369    }
370
371    fn get_send_event_provider(
372        &self,
373        id: &ChannelId,
374    ) -> Option<&dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>> {
375        self.senders
376            .get(id)
377            .filter(|sender| sender.id() == *id)
378            .map(|v| v.as_ref())
379    }
380
381    fn add_send_event_provider(
382        &mut self,
383        send_provider: Box<
384            dyn SendEventProvider<Event, AuxDataProvider, Error = SendProviderError>,
385        >,
386    ) -> bool {
387        let id = send_provider.id();
388        if self.senders.contains_key(&id) {
389            return false;
390        }
391        self.senders.insert(id, send_provider).is_none()
392    }
393}
394
395#[cfg(feature = "std")]
396pub mod stdmod {
397    use super::*;
398    use crate::event_man::{EventReceiver, EventWithAuxData};
399    use crate::events::{EventU16, EventU32, GenericEvent};
400    use crate::params::Params;
401    use std::sync::mpsc::{Receiver, SendError, Sender};
402
403    pub struct MpscEventReceiver<Event: GenericEvent + Send = EventU32> {
404        mpsc_receiver: Receiver<(Event, Option<Params>)>,
405    }
406
407    impl<Event: GenericEvent + Send> MpscEventReceiver<Event> {
408        pub fn new(receiver: Receiver<(Event, Option<Params>)>) -> Self {
409            Self {
410                mpsc_receiver: receiver,
411            }
412        }
413    }
414    impl<Event: GenericEvent + Send> EventReceiver<Event> for MpscEventReceiver<Event> {
415        fn receive(&self) -> Option<EventWithAuxData<Event>> {
416            if let Ok(event_and_data) = self.mpsc_receiver.try_recv() {
417                return Some(event_and_data);
418            }
419            None
420        }
421    }
422
423    pub type MpscEventU32Receiver = MpscEventReceiver<EventU32>;
424    pub type MpscEventU16Receiver = MpscEventReceiver<EventU16>;
425
426    #[derive(Clone)]
427    pub struct MpscEventSendProvider<Event: GenericEvent + Send> {
428        id: u32,
429        sender: Sender<(Event, Option<Params>)>,
430    }
431
432    impl<Event: GenericEvent + Send> MpscEventSendProvider<Event> {
433        pub fn new(id: u32, sender: Sender<(Event, Option<Params>)>) -> Self {
434            Self { id, sender }
435        }
436    }
437
438    impl<Event: GenericEvent + Send> SendEventProvider<Event> for MpscEventSendProvider<Event> {
439        type Error = SendError<(Event, Option<Params>)>;
440
441        fn id(&self) -> u32 {
442            self.id
443        }
444        fn send(&self, event: Event, aux_data: Option<Params>) -> Result<(), Self::Error> {
445            self.sender.send((event, aux_data))
446        }
447    }
448
449    pub type MpscEventU32SendProvider = MpscEventSendProvider<EventU32>;
450    pub type MpscEventU16SendProvider = MpscEventSendProvider<EventU16>;
451}
452
453#[cfg(test)]
454mod tests {
455    use super::*;
456    use crate::event_man::EventManager;
457    use crate::events::{EventU32, GenericEvent, Severity};
458    use crate::params::ParamsRaw;
459    use alloc::boxed::Box;
460    use std::format;
461    use std::sync::mpsc::{channel, Receiver, SendError, Sender};
462
463    #[derive(Clone)]
464    struct MpscEventSenderQueue {
465        id: u32,
466        mpsc_sender: Sender<EventU32WithAuxData>,
467    }
468
469    impl MpscEventSenderQueue {
470        fn new(id: u32, mpsc_sender: Sender<EventU32WithAuxData>) -> Self {
471            Self { id, mpsc_sender }
472        }
473    }
474
475    impl SendEventProvider<EventU32> for MpscEventSenderQueue {
476        type Error = SendError<EventU32WithAuxData>;
477
478        fn id(&self) -> u32 {
479            self.id
480        }
481        fn send(&self, event: EventU32, aux_data: Option<Params>) -> Result<(), Self::Error> {
482            self.mpsc_sender.send((event, aux_data))
483        }
484    }
485
486    fn check_next_event(
487        expected: EventU32,
488        receiver: &Receiver<EventU32WithAuxData>,
489    ) -> Option<Params> {
490        if let Ok(event) = receiver.try_recv() {
491            assert_eq!(event.0, expected);
492            return event.1;
493        }
494        None
495    }
496
497    fn check_handled_event(
498        res: EventRoutingResult<EventU32, Params>,
499        expected: EventU32,
500        expected_num_sent: u32,
501    ) {
502        assert!(matches!(res, EventRoutingResult::Handled { .. }));
503        if let EventRoutingResult::Handled(num_recipients, event, _aux_data) = res {
504            assert_eq!(event, expected);
505            assert_eq!(num_recipients, expected_num_sent);
506        }
507    }
508
509    fn generic_event_man() -> (
510        Sender<EventU32WithAuxData>,
511        EventManager<SendError<EventU32WithAuxData>>,
512    ) {
513        let (event_sender, manager_queue) = channel();
514        let event_man_receiver = MpscEventReceiver::new(manager_queue);
515        (
516            event_sender,
517            EventManager::new(Box::new(event_man_receiver)),
518        )
519    }
520
521    #[test]
522    fn test_basic() {
523        let (event_sender, mut event_man) = generic_event_man();
524        let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap();
525        let event_grp_1_0 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
526        let (single_event_sender, single_event_receiver) = channel();
527        let single_event_listener = MpscEventSenderQueue::new(0, single_event_sender);
528        event_man.subscribe_single(&event_grp_0, single_event_listener.id());
529        event_man.add_sender(single_event_listener);
530        let (group_event_sender_0, group_event_receiver_0) = channel();
531        let group_event_listener = MpscEventSenderQueue {
532            id: 1,
533            mpsc_sender: group_event_sender_0,
534        };
535        event_man.subscribe_group(event_grp_1_0.group_id(), group_event_listener.id());
536        event_man.add_sender(group_event_listener);
537
538        // Test event with one listener
539        event_sender
540            .send((event_grp_0, None))
541            .expect("Sending single error failed");
542        let res = event_man.try_event_handling();
543        assert!(res.is_ok());
544        check_handled_event(res.unwrap(), event_grp_0, 1);
545        check_next_event(event_grp_0, &single_event_receiver);
546
547        // Test event which is sent to all group listeners
548        event_sender
549            .send((event_grp_1_0, None))
550            .expect("Sending group error failed");
551        let res = event_man.try_event_handling();
552        assert!(res.is_ok());
553        check_handled_event(res.unwrap(), event_grp_1_0, 1);
554        check_next_event(event_grp_1_0, &group_event_receiver_0);
555    }
556
557    #[test]
558    fn test_with_basic_aux_data() {
559        let (event_sender, mut event_man) = generic_event_man();
560        let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap();
561        let (single_event_sender, single_event_receiver) = channel();
562        let single_event_listener = MpscEventSenderQueue::new(0, single_event_sender);
563        event_man.subscribe_single(&event_grp_0, single_event_listener.id());
564        event_man.add_sender(single_event_listener);
565        event_sender
566            .send((event_grp_0, Some(Params::Heapless((2_u32, 3_u32).into()))))
567            .expect("Sending group error failed");
568        let res = event_man.try_event_handling();
569        assert!(res.is_ok());
570        check_handled_event(res.unwrap(), event_grp_0, 1);
571        let aux = check_next_event(event_grp_0, &single_event_receiver);
572        assert!(aux.is_some());
573        let aux = aux.unwrap();
574        if let Params::Heapless(ParamsHeapless::Raw(ParamsRaw::U32Pair(pair))) = aux {
575            assert_eq!(pair.0, 2);
576            assert_eq!(pair.1, 3);
577        } else {
578            panic!("{}", format!("Unexpected auxiliary value type {:?}", aux));
579        }
580    }
581
582    /// Test listening for multiple groups
583    #[test]
584    fn test_multi_group() {
585        let (event_sender, mut event_man) = generic_event_man();
586        let res = event_man.try_event_handling();
587        assert!(res.is_ok());
588        let hres = res.unwrap();
589        assert!(matches!(hres, EventRoutingResult::Empty));
590
591        let event_grp_0 = EventU32::new(Severity::INFO, 0, 0).unwrap();
592        let event_grp_1_0 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
593        let (event_grp_0_sender, event_grp_0_receiver) = channel();
594        let event_grp_0_and_1_listener = MpscEventSenderQueue {
595            id: 0,
596            mpsc_sender: event_grp_0_sender,
597        };
598        event_man.subscribe_group(event_grp_0.group_id(), event_grp_0_and_1_listener.id());
599        event_man.subscribe_group(event_grp_1_0.group_id(), event_grp_0_and_1_listener.id());
600        event_man.add_sender(event_grp_0_and_1_listener);
601
602        event_sender
603            .send((event_grp_0, None))
604            .expect("Sending Event Group 0 failed");
605        event_sender
606            .send((event_grp_1_0, None))
607            .expect("Sendign Event Group 1 failed");
608        let res = event_man.try_event_handling();
609        assert!(res.is_ok());
610        check_handled_event(res.unwrap(), event_grp_0, 1);
611        let res = event_man.try_event_handling();
612        assert!(res.is_ok());
613        check_handled_event(res.unwrap(), event_grp_1_0, 1);
614
615        check_next_event(event_grp_0, &event_grp_0_receiver);
616        check_next_event(event_grp_1_0, &event_grp_0_receiver);
617    }
618
619    /// Test listening to the same event from multiple listeners. Also test listening
620    /// to both group and single events from one listener
621    #[test]
622    fn test_listening_to_same_event_and_multi_type() {
623        let (event_sender, mut event_man) = generic_event_man();
624        let event_0 = EventU32::new(Severity::INFO, 0, 5).unwrap();
625        let event_1 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
626        let (event_0_tx_0, event_0_rx_0) = channel();
627        let (event_0_tx_1, event_0_rx_1) = channel();
628        let event_listener_0 = MpscEventSenderQueue {
629            id: 0,
630            mpsc_sender: event_0_tx_0,
631        };
632        let event_listener_1 = MpscEventSenderQueue {
633            id: 1,
634            mpsc_sender: event_0_tx_1,
635        };
636        let event_listener_0_sender_id = event_listener_0.id();
637        event_man.subscribe_single(&event_0, event_listener_0_sender_id);
638        event_man.add_sender(event_listener_0);
639        let event_listener_1_sender_id = event_listener_1.id();
640        event_man.subscribe_single(&event_0, event_listener_1_sender_id);
641        event_man.add_sender(event_listener_1);
642        event_sender
643            .send((event_0, None))
644            .expect("Triggering Event 0 failed");
645        let res = event_man.try_event_handling();
646        assert!(res.is_ok());
647        check_handled_event(res.unwrap(), event_0, 2);
648        check_next_event(event_0, &event_0_rx_0);
649        check_next_event(event_0, &event_0_rx_1);
650        event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id);
651        event_sender
652            .send((event_0, None))
653            .expect("Triggering Event 0 failed");
654        event_sender
655            .send((event_1, None))
656            .expect("Triggering Event 1 failed");
657
658        // 3 Events messages will be sent now
659        let res = event_man.try_event_handling();
660        assert!(res.is_ok());
661        check_handled_event(res.unwrap(), event_0, 2);
662        let res = event_man.try_event_handling();
663        assert!(res.is_ok());
664        check_handled_event(res.unwrap(), event_1, 1);
665        // Both the single event and the group event should arrive now
666        check_next_event(event_0, &event_0_rx_0);
667        check_next_event(event_1, &event_0_rx_0);
668
669        // Do double insertion and then remove duplicates
670        event_man.subscribe_group(event_1.group_id(), event_listener_0_sender_id);
671        event_man.remove_duplicates(&ListenerKey::Group(event_1.group_id()));
672        event_sender
673            .send((event_1, None))
674            .expect("Triggering Event 1 failed");
675        let res = event_man.try_event_handling();
676        assert!(res.is_ok());
677        check_handled_event(res.unwrap(), event_1, 1);
678    }
679
680    #[test]
681    fn test_all_events_listener() {
682        let (event_sender, manager_queue) = channel();
683        let event_man_receiver = MpscEventReceiver::new(manager_queue);
684        let mut event_man: EventManager<SendError<EventU32WithAuxData>> =
685            EventManager::new(Box::new(event_man_receiver));
686        let event_0 = EventU32::new(Severity::INFO, 0, 5).unwrap();
687        let event_1 = EventU32::new(Severity::HIGH, 1, 0).unwrap();
688        let (event_0_tx_0, all_events_rx) = channel();
689        let all_events_listener = MpscEventSenderQueue {
690            id: 0,
691            mpsc_sender: event_0_tx_0,
692        };
693        event_man.subscribe_all(all_events_listener.id());
694        event_man.add_sender(all_events_listener);
695        event_sender
696            .send((event_0, None))
697            .expect("Triggering event 0 failed");
698        event_sender
699            .send((event_1, None))
700            .expect("Triggering event 1 failed");
701        let res = event_man.try_event_handling();
702        assert!(res.is_ok());
703        check_handled_event(res.unwrap(), event_0, 1);
704        let res = event_man.try_event_handling();
705        assert!(res.is_ok());
706        check_handled_event(res.unwrap(), event_1, 1);
707        check_next_event(event_0, &all_events_rx);
708        check_next_event(event_1, &all_events_rx);
709    }
710}
satrs_core/event_man.rs

satrs_core/
event_man.rs
