remove rocket dependencies for manager notifications

src/main.rs

@@ -20,7 +20,8 @@ use notification::Notification;
 use worker::{DeployHandler, Manager};
 
 struct GlobalState {
-    deployer: Manager<DeployHandler>,
+    deployer: Manager<DeployHandler, Notification>,
+    notification: Arc<Notification>,
 }
 
 #[get("/deploy/<id>")]
@@ -33,7 +34,7 @@ async fn deploy(state: &State<GlobalState>, id: u32) -> Value {
 
 #[get("/events")]
 async fn events(state: &State<GlobalState>, end: Shutdown) -> EventStream![] {
-    state.deployer.get_events(end).await
+    state.notification.events(end).await
 }
 
 #[get("/health")]
@@ -44,10 +45,15 @@ async fn healthcheck(state: &State<GlobalState>) -> Value {
 #[rocket::main]
 async fn main() -> Result<(), rocket::Error> {
     let (sender, _receiver) = channel();
-    let mut deployer: Manager<DeployHandler> = Manager::new("deploy", DeployHandler::new(sender));
+
+    let notification = Arc::new(Notification::init());
+    let mut deployer = Manager::new("deploy", DeployHandler::new(sender), notification.clone());
     deployer.launch_workers::<Job>(5);
 
-    let global_state = GlobalState { deployer: deployer };
+    let global_state = GlobalState {
+        deployer,
+        notification,
+    };
 
     let _rocket = rocket::build()
         .mount("/", routes![deploy, events, healthcheck])

src/model/mod.rs

@@ -2,4 +2,4 @@ mod job;
 mod traits;
 
 pub use job::{Job, JobAction};
-pub use traits::{Runner, RunnerStatus, Storer};
+pub use traits::{Notifier, Runner, RunnerStatus, Storer};

src/model/traits.rs

@@ -1,4 +1,4 @@
-use std::marker::Send;
+use std::marker::{Send, Sync};
 
 use crate::error::{MessageError, RunnerError, StorerError};
 use crate::message::Message;
@@ -19,6 +19,10 @@ pub trait Runner: Send + 'static {
         Self: Sized;
 }
 
+pub trait Notifier: Sync + Send + 'static {
+    fn send(&self, data: Message);
+}
+
 pub trait Storer {
     fn insert(&self) -> Result<(), StorerError>;
     fn save(&self) -> Result<(), StorerError>;

src/notification/notification.rs

@@ -2,7 +2,6 @@ use std::sync::{Arc, Mutex};
 
 use rocket::{
     response::stream::{Event, EventStream},
-    serde::Serialize,
     tokio::{
         select,
         sync::{broadcast, broadcast::error::RecvError},
@@ -10,11 +9,14 @@ use rocket::{
     Shutdown,
 };
 
-pub struct Notification<T> {
+use crate::message::Message;
-    sender: Arc<Mutex<broadcast::Sender<T>>>,
+use crate::model::Notifier;
+
+pub struct Notification {
+    sender: Arc<Mutex<broadcast::Sender<Message>>>,
 }
 
-impl<T: Clone + Serialize> Notification<T> {
+impl Notification {
     pub fn init() -> Self {
         let (sender, _) = broadcast::channel(10);
         Self {
@@ -22,7 +24,7 @@ impl<T: Clone + Serialize> Notification<T> {
         }
     }
 
-    pub fn get_sender(&self) -> broadcast::Sender<T> {
+    pub fn get_sender(&self) -> broadcast::Sender<Message> {
         self.sender.lock().unwrap().clone()
     }
 
@@ -44,3 +46,13 @@ impl<T: Clone + Serialize> Notification<T> {
         }
     }
 }
+
+impl Notifier for Notification {
+    fn send(&self, data: Message) {
+        let guard = self.get_sender();
+        match guard.send(data) {
+            Ok(_) => (),
+            Err(e) => eprintln!("error while sending event: {}", e),
+        }
+    }
+}

src/worker/worker.rs

@@ -2,13 +2,10 @@ 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.
 //
@@ -35,22 +32,22 @@ pub enum WorkerStatus {
     Stopped,
 }
 
-struct Worker<T> {
+struct Worker<H, N> {
     id: u32,
     manager: String,
     queue: Arc<Queue<Message>>,
     status: Arc<Mutex<WorkerStatus>>,
-    shared_handler: Arc<T>,
+    shared_handler: Arc<H>,
-    notifier: Arc<Notification<Message>>,
+    notifier: Arc<N>,
 }
 
-impl<T: Handler> Worker<T> {
+impl<H: Handler, N: Notifier> Worker<H, N> {
     fn new(
         id: u32,
         manager: String,
         queue: Arc<Queue<Message>>,
-        shared_handler: Arc<T>,
+        shared_handler: Arc<H>,
-        notifier: Arc<Notification<Message>>,
+        notifier: Arc<N>,
     ) -> Self {
         Self {
             id: id,
@@ -75,7 +72,7 @@ impl<T: Handler> Worker<T> {
     //
     // It tries to parse the incoming `Message` into a `Runner` and runs it.
     // If an `Unknown` error is returned from the handler, the thread is stopped and its status is set to `Failed`.
-    fn launch<U: Runner>(&self) {
+    fn launch<R: Runner>(&self) {
         let queue = self.queue.clone();
         let status = self.status.clone();
         let handler = self.shared_handler.clone();
@@ -93,13 +90,13 @@ impl<T: Handler> Worker<T> {
             };
 
             if message.get_subject() == Subject::StopManager {
-                Worker::<T>::set_status(&status, WorkerStatus::Stopped);
+                Worker::<H, N>::set_status(&status, WorkerStatus::Stopped);
                 break;
             }
 
             drop(guard);
 
-            let mut runner = match U::try_from(message) {
+            let mut runner = match R::try_from(message) {
                 Ok(r) => r,
                 Err(_e) => {
                     eprintln!("unable to parse the incoming message into a runner");
@@ -107,23 +104,21 @@ impl<T: Handler> Worker<T> {
                 }
             };
 
-            Worker::<T>::set_status(&status, WorkerStatus::Running);
+            Worker::<H, N>::set_status(&status, WorkerStatus::Running);
 
             // TODO: the message could be requeued if needed
             if let Err(e) = handler.handle(runner.run()) {
                 match e {
                     HandlerError::Unknown => {
-                        Worker::<T>::set_status(&status, WorkerStatus::Failed);
+                        Worker::<H, N>::set_status(&status, WorkerStatus::Failed);
                         break;
                     }
                     _ => (),
                 }
             }
             // TODO: collect a message from the handler (modify the signature)
-            // TODO: unwrap() on send is useless since it can failed
+            notifier.send(Message::empty());
-            let guard = notifier.get_sender();
+            Worker::<H, N>::set_status(&status, WorkerStatus::Pending);
-            guard.send(Message::empty()).unwrap();
-            Worker::<T>::set_status(&status, WorkerStatus::Pending);
         });
     }
 
@@ -141,18 +136,17 @@ pub enum ManagerStatus {
 }
 
 // Manager is a pool of workers and holds a queue containing `Message`.
-pub struct Manager<T> {
+pub struct Manager<H, N> {
     name: String,
     status: ManagerStatus,
-    workers: Vec<Worker<T>>,
+    workers: Vec<Worker<H, N>>,
     queue: Arc<Queue<Message>>,
-    shared_handler: Arc<T>,
+    shared_handler: Arc<H>,
-    notifier: Arc<Notification<Message>>,
+    notifier: Arc<N>,
 }
 
-impl<T: Handler> Manager<T> {
+impl<H: Handler, N: Notifier> Manager<H, N> {
-    pub fn new(name: &str, shared_handler: T) -> Self {
+    pub fn new(name: &str, shared_handler: H, notifier: Arc<N>) -> Self {
-        let notifier = Arc::new(Notification::<Message>::init());
         Self {
             name: name.to_string(),
             workers: vec![],
@@ -171,25 +165,25 @@ impl<T: Handler> Manager<T> {
     //
     // Example:
     // deployer.launch_workers::<Job>(5)
-    pub fn launch_workers<U: Runner>(&mut self, nb_workers: u32) {
+    pub fn launch_workers<R: Runner>(&mut self, nb_workers: u32) {
         for i in 0..nb_workers {
-            let worker: Worker<T> = Worker::new(
+            let worker: Worker<H, N> = Worker::new(
                 i,
                 self.name.clone(),
                 self.queue.clone(),
                 self.shared_handler.clone(),
                 self.notifier.clone(),
             );
-            worker.launch::<U>();
+            worker.launch::<R>();
             self.workers.push(worker);
         }
 
         self.status = ManagerStatus::Up;
     }
 
-    pub async fn get_events(&self, end: Shutdown) -> EventStream![] {
+    // pub async fn get_events(&self, end: Shutdown) -> EventStream![] {
-        self.notifier.events(end).await
+    //     self.notifier.events(end).await
-    }
+    // }
 
     // subscribe subscribes to a `Receiver` channel and notify all the related workers.
     pub fn subscribe(&self, receiver: Receiver<Message>) {