finished event man tests

.idea/runConfigurations/Test.xml

@@ -7,7 +7,6 @@
     <option name="allFeatures" value="false" />
     <option name="emulateTerminal" value="false" />
     <option name="withSudo" value="false" />
-    <option name="buildTarget" value="REMOTE" />
     <option name="backtrace" value="SHORT" />
     <envs />
     <option name="isRedirectInput" value="false" />

src/core/event_man.rs

@@ -9,11 +9,13 @@ enum ListenerType {
 
 pub trait EventListener {
     type Error;
+
+    fn id(&self) -> u32;
     fn send_to(&mut self, event: Event) -> Result<(), Self::Error>;
 }
 
 struct Listener<E> {
-    _ltype: ListenerType,
+    ltype: ListenerType,
     dest: Box<dyn EventListener<Error = E>>,
 }
 
@@ -26,6 +28,11 @@ pub struct EventManager<E> {
     event_receiver: Box<dyn ReceivesAllEvent>,
 }
 
+pub enum HandlerResult {
+    Empty,
+    Handled(u32, Event),
+}
+
 impl<E> EventManager<E> {
     pub fn new(event_receiver: Box<dyn ReceivesAllEvent>) -> Self {
         EventManager {
@@ -33,7 +40,11 @@ impl<E> EventManager<E> {
             event_receiver,
         }
     }
-    pub fn subcribe_single(&mut self, event: Event, dest: impl EventListener<Error = E> + 'static) {
+    pub fn subscribe_single(
+        &mut self,
+        event: Event,
+        dest: impl EventListener<Error = E> + 'static,
+    ) {
         self.update_listeners(ListenerType::Single(event.raw()), dest);
     }
 
@@ -52,102 +63,251 @@ impl<E> EventManager<E> {
     ) {
         if let std::collections::hash_map::Entry::Vacant(e) = self.listeners.entry(key) {
             e.insert(vec![Listener {
-                _ltype: key,
+                ltype: key,
                 dest: Box::new(dest),
             }]);
         } else {
             let vec = self.listeners.get_mut(&key).unwrap();
+            // To prevent double insertions
+            for entry in vec.iter() {
+                if entry.ltype == key && entry.dest.id() == dest.id() {
+                    return
+                }
+            }
             vec.push(Listener {
-                _ltype: key,
+                ltype: key,
                 dest: Box::new(dest),
             });
         }
     }
 
-    pub fn handle_one_event(&mut self) -> Result<(), E> {
-        let mut status = Ok(());
+    pub fn try_event_handling(&mut self) -> Result<HandlerResult, E> {
+        let mut err_status = None;
+        let mut num_recipients = 0;
+        let mut send_handler = |event, llist: &mut Vec<Listener<E>>| {
+            for listener in llist.iter_mut() {
+                if let Err(e) = listener.dest.send_to(event) {
+                    err_status = Some(Err(e));
+                } else {
+                    num_recipients += 1;
+                }
+            }
+        };
         if let Some(event) = self.event_receiver.receive() {
-            for (ltype, listener_list) in self.listeners.iter_mut() {
-                match ltype {
-                    ListenerType::Single(raw_event) => {
-                        if event.raw() == *raw_event {
-                            for listener in listener_list.iter_mut() {
-                                if let Err(e) = listener.dest.send_to(event) {
-                                    status = Err(e);
+            let single_key = ListenerType::Single(event.raw());
+            if self.listeners.contains_key(&single_key) {
+                send_handler(event, self.listeners.get_mut(&single_key).unwrap());
             }
+            let group_key = ListenerType::Group(event.group_id());
+            if self.listeners.contains_key(&group_key) {
+                send_handler(event, self.listeners.get_mut(&group_key).unwrap());
             }
+            if let Some(err) = err_status {
+                return err;
             }
+            return Ok(HandlerResult::Handled(num_recipients, event));
         }
-                    ListenerType::Group(group_id) => {
-                        if event.group_id() == *group_id {
-                            for listener in listener_list.iter_mut() {
-                                if let Err(e) = listener.dest.send_to(event) {
-                                    status = Err(e);
-                                }
-                            }
-                        }
-                    }
-                }
-            }
-        }
-        status
+        Ok(HandlerResult::Empty)
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use super::{EventListener, ReceivesAllEvent};
+    use super::{EventListener, HandlerResult, ReceivesAllEvent};
     use crate::core::event_man::EventManager;
     use crate::core::events::{Event, Severity};
-    use std::fmt::Error;
     use std::sync::mpsc::{channel, Receiver, SendError, Sender};
+    use std::thread;
+    use std::time::Duration;
 
     struct EventReceiver {
         mpsc_receiver: Receiver<Event>,
     }
     impl ReceivesAllEvent for EventReceiver {
         fn receive(&mut self) -> Option<Event> {
-            self.mpsc_receiver.recv().ok()
+            self.mpsc_receiver.try_recv().ok()
         }
     }
 
+    #[derive(Clone)]
     struct MpscEventSenderQueue {
+        id: u32,
         mpsc_sender: Sender<Event>,
     }
 
     impl EventListener for MpscEventSenderQueue {
         type Error = SendError<Event>;
 
+        fn id(&self) -> u32 {
+            self.id
+        }
         fn send_to(&mut self, event: Event) -> Result<(), Self::Error> {
             self.mpsc_sender.send(event)
         }
     }
 
+    fn check_next_event(expected: Event, receiver: &Receiver<Event>) {
+        for _ in 0..5 {
+            if let Ok(event) = receiver.try_recv() {
+                assert_eq!(event, expected);
+                break;
+            }
+            thread::sleep(Duration::from_millis(1));
+        }
+    }
+
+    fn check_handled_event(res: HandlerResult, expected: Event, expected_num_sent: u32) {
+        assert!(matches!(res, HandlerResult::Handled { .. }));
+        match res {
+            HandlerResult::Handled(num_recipients, event) => {
+                assert_eq!(event, expected);
+                assert_eq!(num_recipients, expected_num_sent);
+            }
+            _ => (),
+        }
+    }
+
     #[test]
-    fn basic_test() {
+    fn test_basic() {
         let (event_sender, manager_queue) = channel();
         let event_man_receiver = EventReceiver {
             mpsc_receiver: manager_queue,
         };
         let mut event_man: EventManager<SendError<Event>> =
             EventManager::new(Box::new(event_man_receiver));
-        let sender_0 = event_sender.clone();
         let event_grp_0 = Event::new(Severity::INFO, 0, 0).unwrap();
         let event_grp_1_0 = Event::new(Severity::HIGH, 1, 0).unwrap();
-        let event_grp_1_1 = Event::new(Severity::MEDIUM, 1, 32).unwrap();
-        let event_grp_1_2 = Event::new(Severity::LOW, 1, 43).unwrap();
-        let event_grp_2 = Event::new(Severity::HIGH, 32, 0).unwrap();
         let (single_event_sender, single_event_receiver) = channel();
         let single_event_listener = MpscEventSenderQueue {
+            id: 0,
             mpsc_sender: single_event_sender,
         };
-        let (group_event_sender, group_event_receiver) = channel();
+        event_man.subscribe_single(event_grp_0, single_event_listener);
+        let (group_event_sender_0, group_event_receiver_0) = channel();
         let group_event_listener = MpscEventSenderQueue {
-            mpsc_sender: group_event_sender,
+            id: 1,
+            mpsc_sender: group_event_sender_0,
         };
         event_man.subscribe_group(event_grp_1_0.group_id(), group_event_listener);
-        event_sender.send(event_grp_0).expect("Send error occured");
-        // let event = single_event_receiver.recv().expect("Error receiving event");
-        // println!("{:?}", event);
+
+        // Test event with one listener
+        event_sender
+            .send(event_grp_0)
+            .expect("Sending single error failed");
+        let res = event_man.try_event_handling();
+        assert!(res.is_ok());
+        check_handled_event(res.unwrap(), 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(event_grp_1_0)
+            .expect("Sending group error failed");
+        let res = event_man.try_event_handling();
+        assert!(res.is_ok());
+        check_handled_event(res.unwrap(), event_grp_1_0, 1);
+        check_next_event(event_grp_1_0, &group_event_receiver_0);
+    }
+
+    /// Test listening for multiple groups
+    #[test]
+    fn test_multi_group() {
+        let (event_sender, manager_queue) = channel();
+        let event_man_receiver = EventReceiver {
+            mpsc_receiver: manager_queue,
+        };
+        let mut event_man: EventManager<SendError<Event>> =
+            EventManager::new(Box::new(event_man_receiver));
+        let res = event_man.try_event_handling();
+        assert!(res.is_ok());
+        let hres = res.unwrap();
+        assert!(matches!(hres, HandlerResult::Empty));
+
+        let event_grp_0 = Event::new(Severity::INFO, 0, 0).unwrap();
+        let event_grp_1_0 = Event::new(Severity::HIGH, 1, 0).unwrap();
+        let (event_grp_0_sender, event_grp_0_receiver) = channel();
+        let event_grp_0_and_1_listener = MpscEventSenderQueue {
+            id: 0,
+            mpsc_sender: event_grp_0_sender,
+        };
+        event_man.subscribe_group(event_grp_0.group_id(), event_grp_0_and_1_listener.clone());
+        event_man.subscribe_group(event_grp_1_0.group_id(), event_grp_0_and_1_listener);
+
+        event_sender
+            .send(event_grp_0)
+            .expect("Sending Event Group 0 failed");
+        event_sender
+            .send(event_grp_1_0)
+            .expect("Sendign Event Group 1 failed");
+        let res = event_man.try_event_handling();
+        assert!(res.is_ok());
+        check_handled_event(res.unwrap(), event_grp_0, 1);
+        let res = event_man.try_event_handling();
+        assert!(res.is_ok());
+        check_handled_event(res.unwrap(), 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 (event_sender, manager_queue) = channel();
+        let event_man_receiver = EventReceiver {
+            mpsc_receiver: manager_queue,
+        };
+        let mut event_man: EventManager<SendError<Event>> =
+            EventManager::new(Box::new(event_man_receiver));
+        let event_0 = Event::new(Severity::INFO, 0, 5).unwrap();
+        let event_1 = Event::new(Severity::HIGH, 1, 0).unwrap();
+        let (event_0_tx_0, event_0_rx_0) = channel();
+        let (event_0_tx_1, event_0_rx_1) = channel();
+        let event_listener_0 = MpscEventSenderQueue {
+            id: 0,
+            mpsc_sender: event_0_tx_0,
+        };
+        let event_listener_1 = MpscEventSenderQueue {
+            id: 1,
+            mpsc_sender: event_0_tx_1,
+        };
+        event_man.subscribe_single(event_0, event_listener_0.clone());
+        event_man.subscribe_single(event_0, event_listener_1);
+        event_sender
+            .send(event_0)
+            .expect("Triggering Event 0 failed");
+        let res = event_man.try_event_handling();
+        assert!(res.is_ok());
+        check_handled_event(res.unwrap(), 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.clone());
+        event_sender
+            .send(event_0)
+            .expect("Triggering Event 0 failed");
+        event_sender
+            .send(event_1)
+            .expect("Triggering Event 1 failed");
+
+        // 3 Events messages will be sent now
+        let res = event_man.try_event_handling();
+        assert!(res.is_ok());
+        check_handled_event(res.unwrap(), event_0, 2);
+        let res = event_man.try_event_handling();
+        assert!(res.is_ok());
+        check_handled_event(res.unwrap(), 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);
+
+        // Double insertion should be detected, result should remain the same
+        event_man.subscribe_group(event_1.group_id(), event_listener_0);
+        event_sender
+            .send(event_1)
+            .expect("Triggering Event 1 failed");
+        let res = event_man.try_event_handling();
+        assert!(res.is_ok());
+        check_handled_event(res.unwrap(), event_1, 1);
     }
 }

src/core/events.rs

@@ -26,7 +26,7 @@ impl TryFrom<u8> for Severity {
     }
 }
 
-#[derive(Copy, Clone, Debug)]
+#[derive(Copy, Clone, Debug, PartialEq)]
 pub struct Event {
     severity: Severity,
     group_id: GroupId,