Graceful shutdown

Cargo.toml

@@ -24,8 +24,8 @@ actix = "0.7"
 actix-web = "0.7"
 bb8 = "0.3"
 bb8-redis = "0.3"
-config = "0.9"
 chrono = "0.4"
+config = "0.9"
 dotenv = "0.13"
 failure = "0.1"
 futures = "0.1"
@@ -41,11 +41,11 @@ rust-crypto = "0.2"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 slog = "2.4"
-slog-envlogger = "2.1"
-slog-stdlog = "3.0"
 slog-async = "2.3"
+slog-envlogger = "2.1"
 slog-json = "2.3"
 slog-scope = "4.1"
+slog-stdlog = "3.0"
 smart-default = "0.5"
 tokio = "0.1"
 toml = "0.4"

docker-compose.yml

@@ -1,4 +1,4 @@
-version: 2
+version: "2"
 
 services:
   coturn:

jason/src/utils/event_listener.rs

@@ -51,7 +51,7 @@ impl<T: Deref<Target = EventTarget>, A: FromWasmAbi + 'static>
     where
         F: FnOnce(A) + 'static,
     {
-        let closure: Closure<FnMut(A)> = Closure::once(closure);
+        let closure: Closure<dyn FnMut(A)> = Closure::once(closure);
 
         target.add_event_listener_with_callback(
             event_name,

proto/client-api/src/lib.rs

@@ -93,8 +93,8 @@ pub struct Track {
     pub media_type: MediaType,
 }
 
-/// Representation of [RTCIceServer][1] (item of `iceServers` field
-/// from [RTCConfiguration][2]).
+/// Representation of [`RTCIceServer`][1] (item of `iceServers` field
+/// from [`RTCConfiguration`][2]).
 ///
 /// [1]: https://developer.mozilla.org/en-US/docs/Web/API/RTCIceServer
 /// [2]: https://developer.mozilla.org/en-US/docs/Web/API/RTCConfiguration

src/api/client/rpc_connection.rs

@@ -1,6 +1,6 @@
 //! [`RpcConnection`] with related messages.
 
-use core::fmt;
+use std::fmt;
 
 use actix::Message;
 use futures::Future;

src/api/client/server.rs

@@ -97,15 +97,18 @@ pub fn run(rooms: RoomsRepository, config: Conf) {
     .unwrap()
     .start();
 
-    info!("Started HTTP server on 0.0.0.0:8080");
+    info!("Started HTTP server on {:?}", server_addr);
 }
 
 #[cfg(test)]
 mod test {
     use std::{ops::Add, thread, time::Duration};
 
     use actix::Arbiter;
-    use actix_web::{http, test, App};
+    use actix_web::{
+        actix::{actors::signal, System},
+        http, test, App,
+    };
     use futures::Stream;
 
     use crate::{
@@ -132,13 +135,18 @@ mod test {
                 ice_user: None
             },
         };
+
+        let process_signals =
+            System::current().registry().get::<signal::ProcessSignals>();
+
         let room = Arbiter::start(move |_| {
             Room::new(
                 1,
                 members,
                 create_peers(1, 2),
                 conf.reconnect_timeout,
                 new_turn_auth_service_mock(),
+                process_signals,
             )
         });
         let rooms = hashmap! {1 => room};

src/main.rs

@@ -9,7 +9,7 @@ pub mod media;
 pub mod signalling;
 pub mod turn;
 
-use actix::prelude::*;
+use actix::{actors::signal, prelude::Arbiter, System};
 use dotenv::dotenv;
 use log::prelude::*;
 
@@ -26,9 +26,7 @@ fn main() {
     let logger = log::new_dual_logger(std::io::stdout(), std::io::stderr());
     let _scope_guard = slog_scope::set_global_logger(logger);
     slog_stdlog::init().unwrap();
-
     let sys = System::new("medea");
-
     let config = Conf::parse().unwrap();
 
     info!("{:?}", config);
@@ -39,6 +37,9 @@ fn main() {
     };
     let peers = create_peers(1, 2);
 
+    let process_signals =
+        System::current().registry().get::<signal::ProcessSignals>();
+
     let turn_auth_service =
         new_turn_auth_service(&config).expect("Unable to start turn service");
     let room = Room::new(
@@ -47,6 +48,7 @@ fn main() {
         peers,
         config.rpc.reconnect_timeout,
         turn_auth_service,
+        process_signals,
     );
     let room = Arbiter::start(move |_| room);
     let rooms = hashmap! {1 => room};

src/signalling/room.rs

@@ -1,11 +1,12 @@
 //! Room definitions and implementations. Room is responsible for media
 //! connection establishment between concrete [`Member`]s.
 
-use std::time::Duration;
+use std::{fmt, time::Duration};
 
 use actix::{
-    fut::wrap_future, Actor, ActorFuture, AsyncContext, Context, Handler,
-    Message,
+    actors::signal::{self, ProcessSignals, Subscribe},
+    fut::wrap_future,
+    Actor, ActorFuture, Addr, AsyncContext, Context, Handler, Message,
 };
 use failure::Fail;
 use futures::future;
@@ -37,7 +38,31 @@ pub type Id = u64;
 pub type ActFuture<I, E> =
     Box<dyn ActorFuture<Actor = Room, Item = I, Error = E>>;
 
-#[derive(Fail, Debug)]
+macro_rules! important_quit_signals {
+    ($msg: expr, $action: expr) => {
+        match $msg {
+            signal::SignalType::Int => {
+                error!("SIGINT received by Room, exiting");
+                $action();
+            }
+            signal::SignalType::Hup => {
+                error!("SIGHUP received by Room, reloading");
+                $action();
+            }
+            signal::SignalType::Term => {
+                error!("SIGTERM received by Room, stopping");
+                $action();
+            }
+            signal::SignalType::Quit => {
+                error!("SIGQUIT received by Room, exiting");
+                $action();
+            }
+            _ => (),
+        }
+    };
+}
+
+#[derive(Debug, Fail)]
 #[allow(clippy::module_name_repetitions)]
 pub enum RoomError {
     #[fail(display = "Couldn't find Peer with [id = {}]", _0)]
@@ -65,7 +90,6 @@ impl From<PeerStateError> for RoomError {
 }
 
 /// Media server room with its [`Member`]s.
-#[derive(Debug)]
 pub struct Room {
     id: Id,
 
@@ -74,6 +98,9 @@ pub struct Room {
 
     /// [`Peer`]s of [`Member`]s in this [`Room`].
     peers: PeerRepository,
+
+    /// Actix addr of [`ProcessSignals`]
+    process_signals: Addr<ProcessSignals>,
 }
 
 impl Room {
@@ -84,6 +111,7 @@ impl Room {
         peers: HashMap<PeerId, PeerStateMachine>,
         reconnect_timeout: Duration,
         turn: Box<dyn TurnAuthService>,
+        process_signals: Addr<ProcessSignals>,
     ) -> Self {
         Self {
             id,
@@ -94,6 +122,7 @@ impl Room {
                 turn,
                 reconnect_timeout,
             ),
+            process_signals,
         }
     }
 
@@ -261,11 +290,26 @@ impl Room {
     }
 }
 
+impl fmt::Debug for Room {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            "Room {{ id: {:?}, participants: {:?}, peers: {:?} }}",
+            self.id, self.participants, self.peers
+        )
+    }
+}
+
 /// [`Actor`] implementation that provides an ergonomic way
 /// to interact with [`Room`].
 // TODO: close connections on signal (gracefull shutdown)
 impl Actor for Room {
     type Context = Context<Self>;
+
+    fn started(&mut self, ctx: &mut Self::Context) {
+        self.process_signals
+            .do_send(Subscribe(ctx.address().recipient()));
+    }
 }
 
 impl Handler<Authorize> for Room {
@@ -429,6 +473,15 @@ impl Handler<CloseRoom> for Room {
     }
 }
 
+// Close room on `SIGINT`, `SIGTERM`, `SIGQUIT` signals.
+impl Handler<signal::Signal> for Room {
+    type Result = ();
+
+    fn handle(&mut self, msg: signal::Signal, ctx: &mut Self::Context) {
+        important_quit_signals!(msg.0, || { ctx.notify(CloseRoom {}) });
+    }
+}
+
 impl Handler<RpcConnectionClosed> for Room {
     type Result = ();
 
@@ -474,13 +527,18 @@ mod test {
                 ice_user: None
             },
         };
+
+        let process_signals =
+            System::current().registry().get::<signal::ProcessSignals>();
+
         Arbiter::start(move |_| {
             Room::new(
                 1,
                 members,
                 create_peers(1, 2),
                 Duration::from_secs(10),
                 new_turn_auth_service_mock(),
+                process_signals,
             )
         })
     }

src/signalling/room_repo.rs

@@ -1,10 +1,10 @@
 //! Repository that stores [`Room`]s addresses.
 
+use std::sync::{Arc, Mutex};
+
 use actix::Addr;
 use hashbrown::HashMap;
 
-use std::sync::{Arc, Mutex};
-
 use crate::signalling::{Room, RoomId};
 
 /// Repository that stores [`Room`]s addresses.

src/turn/service.rs

@@ -46,7 +46,7 @@ impl TurnAuthService for Addr<Service> {
         member_id: u64,
         room_id: RoomId,
         policy: UnreachablePolicy,
-    ) -> Box<Future<Item = IceUser, Error = TurnServiceErr>> {
+    ) -> Box<dyn Future<Item = IceUser, Error = TurnServiceErr>> {
         Box::new(
             self.send(CreateIceUser {
                 member_id,
@@ -69,7 +69,7 @@ impl TurnAuthService for Addr<Service> {
     fn delete(
         &self,
         users: Vec<IceUser>,
-    ) -> Box<Future<Item = (), Error = TurnServiceErr>> {
+    ) -> Box<dyn Future<Item = (), Error = TurnServiceErr>> {
         // leave only non static users
         let users: Vec<IceUser> =
             users.into_iter().filter(|u| !u.is_static()).collect();