impl notification + add Rocket endpoints

src/worker/worker.rs

@@ -2,10 +2,13 @@ use std::collections::VecDeque;
 use std::sync::{mpsc::Receiver, Arc, Condvar, Mutex};
 use std::{thread, time};
 
+use rocket::{response::stream::EventStream, Shutdown};
+
 use super::handler::Handler;
 use crate::error::HandlerError;
 use crate::message::{Message, Subject};
 use crate::model::*;
+use crate::notification::Notification;
 
 // Queue is a simple queue data structure.
 //
@@ -38,16 +41,24 @@ struct Worker<T> {
     queue: Arc<Queue<Message>>,
     status: Arc<Mutex<WorkerStatus>>,
     shared_handler: Arc<T>,
+    notifier: Arc<Notification<Message>>,
 }
 
 impl<T: Handler> Worker<T> {
-    fn new(id: u32, manager: String, queue: Arc<Queue<Message>>, shared_handler: Arc<T>) -> Self {
+    fn new(
+        id: u32,
+        manager: String,
+        queue: Arc<Queue<Message>>,
+        shared_handler: Arc<T>,
+        notifier: Arc<Notification<Message>>,
+    ) -> Self {
         Self {
             id: id,
             manager: manager,
             queue: queue,
             status: Arc::new(Mutex::new(WorkerStatus::Pending)),
             shared_handler: shared_handler,
+            notifier: notifier,
         }
     }
 
@@ -68,6 +79,7 @@ impl<T: Handler> Worker<T> {
         let queue = self.queue.clone();
         let status = self.status.clone();
         let handler = self.shared_handler.clone();
+        let notifier = self.notifier.clone();
 
         thread::spawn(move || loop {
             let mut guard = queue.content.lock().unwrap();
@@ -107,6 +119,10 @@ impl<T: Handler> Worker<T> {
                     _ => (),
                 }
             }
+            // TODO: collect a message from the handler (modify the signature)
+            // TODO: unwrap() on send is useless since it can failed
+            let guard = notifier.get_sender();
+            guard.send(Message::empty()).unwrap();
             Worker::<T>::set_status(&status, WorkerStatus::Pending);
         });
     }
@@ -131,16 +147,19 @@ pub struct Manager<T> {
     workers: Vec<Worker<T>>,
     queue: Arc<Queue<Message>>,
     shared_handler: Arc<T>,
+    notifier: Arc<Notification<Message>>,
 }
 
 impl<T: Handler> Manager<T> {
     pub fn new(name: &str, shared_handler: T) -> Self {
+        let notifier = Arc::new(Notification::<Message>::init());
         Self {
             name: name.to_string(),
             workers: vec![],
             status: ManagerStatus::Down,
             queue: Arc::new(Queue::new()),
             shared_handler: Arc::new(shared_handler),
+            notifier: notifier,
         }
     }
 
@@ -159,6 +178,7 @@ impl<T: Handler> Manager<T> {
                 self.name.clone(),
                 self.queue.clone(),
                 self.shared_handler.clone(),
+                self.notifier.clone(),
             );
             worker.launch::<U>();
             self.workers.push(worker);
@@ -167,6 +187,10 @@ impl<T: Handler> Manager<T> {
         self.status = ManagerStatus::Up;
     }
 
+    pub async fn get_events(&self, end: Shutdown) -> EventStream![] {
+        self.notifier.events(end).await
+    }
+
     // subscribe subscribes to a `Receiver` channel and notify all the related workers.
     pub fn subscribe(&self, receiver: Receiver<Message>) {
         let queue = self.queue.clone();
@@ -213,7 +237,7 @@ impl<T: Handler> Manager<T> {
         self.status = ManagerStatus::Stopping;
 
         let wait = time::Duration::from_millis(100);
-        while !self.healthcheck(WorkerStatus::Stopped) {
+        while !self.check_status(WorkerStatus::Stopped) {
             thread::sleep(wait);
         }
 
@@ -221,7 +245,16 @@ impl<T: Handler> Manager<T> {
     }
 
     // healthcheck checks the status of all workers.
-    pub fn healthcheck(&self, target: WorkerStatus) -> bool {
+    pub fn healthcheck(&self) -> bool {
+        for w in &self.workers {
+            if w.get_status() == WorkerStatus::Stopped || w.get_status() == WorkerStatus::Failed {
+                return false;
+            }
+        }
+        true
+    }
+
+    fn check_status(&self, target: WorkerStatus) -> bool {
         for w in &self.workers {
             if w.get_status() != target {
                 return false;