imp(timer): feature parity for kqueue

Author: omarabid

src/timer/kqueue.rs

@@ -4,6 +4,7 @@
 // https://www.apache.org/licenses/LICENSE-2.0>. This file may not be copied, modified, or distributed
 // except according to those terms.
 
+use super::TimerSignal;
 use crate::utils::check_err;
 use crate::utils::get_time_range;
 use crate::Result;
@@ -28,65 +29,67 @@ use std::{
 #[derive(Debug, Default)]
 pub struct Timer {
     /// A vector to store listeners (mpsc::Sender)
-    txs: Arc<Mutex<Vec<Sender<u64>>>>,
+    txs: Arc<Mutex<Vec<Sender<TimerSignal>>>>,
 
     /// Thread handle
     pub handle: Option<JoinHandle<Result<()>>>,
 }
 
 impl Timer {
     /// Initialize Timer and run a thread to send events to attached listeners
-    pub fn initialize(self, accumulation_cycle: Duration) -> Result<Self> {
-        let txs = Arc::clone(&self.txs);
+    pub fn initialize(cycle: Duration) -> Result<Self> {
+        let txs = Arc::new(Mutex::new(Vec::new()));
 
         // Add Default tx
-        let (tx, _rx): (Sender<u64>, Receiver<u64>) = channel();
+        let (tx, _rx): (Sender<TimerSignal>, Receiver<TimerSignal>) = channel();
         txs.lock()?.push(tx);
 
         let kqueue = kqueue()?;
 
-        let handle = Some(thread::spawn(move || {
-            // Wait for initial expiration
-            let initial_event =
-                Timer::register_initial_expiration(kqueue, Duration::from_millis(0))?;
-            Timer::wait_event(kqueue, [initial_event].as_mut_ptr())?;
-
-            // Register loop event
-            let loop_event = Timer::register_loop_expiration(kqueue, accumulation_cycle)?;
-
-            // Loop 10s
-            loop {
-                // Exit thread if there are no listeners
-                if txs.lock()?.len() == 0 {
-                    // TODO: should close file descriptors?
-                    return Ok(());
-                }
+        let handle = Some({
+            let txs = txs.clone();
+            thread::spawn(move || {
+                // Wait for initial expiration
+                let initial_event = Timer::register_initial_expiration(kqueue)?;
+                Timer::wait_event(kqueue, [initial_event].as_mut_ptr())?;
+
+                // Register loop event
+                let loop_event = Timer::register_loop_expiration(kqueue, cycle)?;
+
+                // Loop 10s
+                loop {
+                    // Exit thread if there are no listeners
+                    if txs.lock()?.len() == 0 {
+                        // TODO: should close file descriptors?
+                        return Ok(());
+                    }
 
-                // Get current time
-                let from = get_time_range(0)?.from;
+                    // Get current time
+                    let from = TimerSignal::NextSnapshot(get_time_range(0)?.from);
 
-                // Iterate through Senders
-                txs.lock()?.iter().for_each(|tx| {
-                    // Send event to attached Sender
-                    match tx.send(from) {
-                        Ok(_) => {}
-                        Err(_) => {}
-                    }
-                });
+                    // Iterate through Senders
+                    txs.lock()?.iter().for_each(|tx| {
+                        // Send event to attached Sender
+                        match tx.send(from) {
+                            Ok(_) => {}
+                            Err(_) => {}
+                        }
+                    });
 
-                // Wait 10s
-                Timer::wait_event(kqueue, [loop_event].as_mut_ptr())?;
-            }
-        }));
+                    // Wait 10s
+                    Timer::wait_event(kqueue, [loop_event].as_mut_ptr())?;
+                }
+            })
+        });
 
-        Ok(Self { handle, ..self })
+        Ok(Self { handle, txs })
     }
 
     /// Attach an mpsc::Sender to Timer
     ///
     /// Timer will dispatch an event with the timestamp of the current instant,
     /// every 10th second to all attached senders
-    pub fn attach_listener(&mut self, tx: Sender<u64>) -> Result<()> {
+    pub fn attach_listener(&mut self, tx: Sender<TimerSignal>) -> Result<()> {
         // Push Sender to a Vector of Sender(s)
         let txs = Arc::clone(&self.txs);
         txs.lock()?.push(tx);