Tear out warp, use actix instead

Cargo.toml

@@ -7,10 +7,11 @@ edition = "2018"
 [dependencies]
     rand = "0.6.5"
     typed-html = "0.1.1"
-    warp = "0.1.13"
+    actix-web = "0.7.18"
+    actix-net = "0.2.6"
+    actix = "0.7.4"
     futures = "0.1.25"
     log = "0.4.6"
     simplelog = "0.5.3"
     crossbeam-channel = "0.3.8"
     ctrlc = "3.1.1"
-    tokio = "0.1.15"

src/adaptors.rs

@@ -26,14 +26,13 @@ impl FakeAdaptor {
     }
 
     fn _run(numchannels: u8, sender: Sender<TallyState>) {
-        println!("Hello, I have {} channels", numchannels);
         loop {
             let state = TallyState {
                 on_air: Channel(Some(rand::thread_rng().gen_range(0, numchannels))),
                 preview: Channel(Some(rand::thread_rng().gen_range(0, numchannels))),
             };
             if let Err(_) = sender.send(state) {
-                log::info!("Fakeadaptor shutting down!");
+                log::info!("Fakeadaptor shutting down");
                 break;
             };
             thread::sleep(Duration::new(rand::thread_rng().gen_range(0, 5), 0));
@@ -49,7 +48,7 @@ impl Waitable for FakeAdaptor {
 
 impl Adaptor for FakeAdaptor {
     fn run(&mut self) -> Receiver<TallyState> {
-        println!("arse");
+        log::info!("Fakeadaptor starting");
         let (tx, rx) = crossbeam_channel::unbounded::<TallyState>();
         let thread_numchannels = self.numchannels.clone();
         self.thread = Some(thread::spawn(move || {

src/drivers.rs

@@ -21,6 +21,7 @@ impl Waitable for FakeDriver {
 }
 impl Driver for FakeDriver {
     fn run(&mut self) -> Sender<TallyState> {
+        log::info!("Fakedriver starting");
         let (tx, rx) = crossbeam_channel::bounded::<TallyState>(1);
         self.thread = Some(thread::spawn(move || {
             FakeDriver::_run(rx);
@@ -40,7 +41,7 @@ impl FakeDriver {
                 println!("{}", state);
             } else {
                 // channel has been disconnected, shutdown.
-                log::info!("Fakedriver shutting down...");
+                log::info!("Fakedriver shutting down");
                 return;
             }
         }

src/web.rs

@@ -1,13 +1,11 @@
 use crate::TallyState;
-use futures::stream;
-use futures::sync::mpsc::Receiver;
-use futures::sync::oneshot;
-use std::sync::{Arc, Mutex};
+use actix::prelude::*;
+use actix_web::{server, App, HttpResponse};
+use futures::future::Future;
+use std::sync::mpsc;
 use std::thread;
 use typed_html::dom::DOMTree;
 use typed_html::html;
-use warp::filters::ws::Message;
-use warp::{Filter, Future, Stream};
 
 fn build_template() -> DOMTree<String> {
     html!(
@@ -24,12 +22,12 @@ fn build_template() -> DOMTree<String> {
 }
 
 pub struct WebServer {
-    shutdown_chan: oneshot::Sender<()>,
-    shutdown_mutex: Arc<Mutex<()>>,
+    addr: Addr<actix_net::server::Server>,
+    shutdown_rx: mpsc::Receiver<()>,
 }
 
 impl WebServer {
-    pub fn run(addr: std::net::SocketAddr, tally_rx: Receiver<TallyState>) -> WebServer {
+    pub fn run(addr: std::net::SocketAddr) -> WebServer {
         // todo: not have the damn js inline in the func
         const JS: &'static str = r#"
 var sock = new WebSocket("ws://" + window.location.host + "/ws");
@@ -39,46 +37,50 @@ sock.onmessage = function (e) {
 sock.onopen = function (e) {
     sock.send("Hello!");
 }"#;
-        let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
-
-        let shutdown_mutex = Arc::new(Mutex::new(()));
-        let thread_shutdown_mutex = Arc::clone(&shutdown_mutex);
+        let (shutdown_tx, shutdown_rx) = mpsc::channel();
+        let (tx, rx) = mpsc::channel();
         thread::spawn(move || {
-            // N.B! Despite the fact we never use the value inside the mutex (it's () after all..) we need to bind
-            // it to a variable so that it lives until the end of the thread, else it would unlock immediately.
-            let _lock = thread_shutdown_mutex.lock().unwrap();
-
-            let websocket = warp::path("ws").and(warp::ws2()).map(|ws: warp::ws::Ws2| {
-                ws.on_upgrade(|websocket| {
-                    let (tx, _) = websocket.split();
-                    //stream::once::<Message, _>(Ok(Message::text("Hello")))
-                    tally_rx
-                        .map(|state| Message::text(format!("{}", state)))
-                        .forward(tx)
-                        .map(|_| ())
-                        .map_err(|e| {
-                            log::error!("Websocket error: {}", e);
+            let sys = actix::System::new("test");
+            let addr = server::new(|| {
+                App::new()
+                    .resource("/", |r| {
+                        r.f(|_r| {
+                            HttpResponse::Ok()
+                                .content_type("text/html")
+                                .body(build_template().to_string())
                         })
-                })
-            });
-            let script = warp::path("script.js").map(|| JS);
-            let home = warp::path::end().map(|| warp::reply::html(build_template().to_string()));
-            let routes = warp::get2().and(websocket.or(script).or(home));
+                    })
+                    .resource("script.js", |r| {
+                        r.f(|_r| HttpResponse::Ok().content_type("text/javascript").body(JS))
+                    })
+                //.resource("/ws", )
+            })
+            .bind(addr)
+            .expect("Error binding to address.")
+            .start();
 
-            let (_addr, server) =
-                warp::serve(routes).bind_with_graceful_shutdown(addr, shutdown_rx);
-            tokio::run(server);
+            tx.send(addr).unwrap();
+
+            sys.run();
+            shutdown_tx.send(()).unwrap(); // Notify that we have shutdown
         });
+
+        let addr = rx.recv().unwrap();
+
         WebServer {
-            shutdown_chan: shutdown_tx,
-            shutdown_mutex,
+            addr: addr,
+            shutdown_rx: shutdown_rx,
         }
     }
 
     pub fn shutdown(self) {
         log::debug!("Web server shutting down...");
-        self.shutdown_chan.send(()).unwrap();
-        let _lock = self.shutdown_mutex.lock().unwrap();
+        self.addr
+            .send(actix_web::server::StopServer { graceful: true })
+            .wait()
+            .unwrap()
+            .unwrap();
+        let _ = self.shutdown_rx.recv().unwrap(); // wait for webserver shutdown
         log::info!("Web server shutdown");
     }
 }