Graceful shutdown

config.toml

@@ -17,7 +17,7 @@
 # heartbeat messages received.
 #
 # Default:
-#   idle_timeout = "10s"
+   idle_timeout = "100s"
 
 # Duration, after which the server deletes the client session if
 # the remote RPC client does not reconnect after it is idle.

docker-compose.yml

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

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

@@ -5,9 +5,10 @@ use core::fmt;
 use actix::Message;
 use futures::Future;
 use macro_attr::*;
-use medea_client_api_proto::{Command, Event};
 use newtype_derive::NewtypeFrom;
 
+use medea_client_api_proto::{Command, Event};
+
 use crate::api::control::MemberId;
 
 macro_attr! {
@@ -97,6 +98,7 @@ pub mod test {
         System,
     };
     use futures::future::Future;
+
     use medea_client_api_proto::{Command, Event, IceCandidate};
 
     use crate::{

src/api/client/server.rs

@@ -97,14 +97,14 @@ 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::{Arbiter, System};
     use actix_web::{http, test, App};
     use futures::Stream;
 
@@ -132,13 +132,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/api/client/session.rs

@@ -8,6 +8,7 @@ use actix::{
 };
 use actix_web::ws::{self, CloseReason, WebsocketContext};
 use futures::future::Future;
+
 use medea_client_api_proto::{ClientMsg, ServerMsg};
 
 use crate::{

src/main.rs

@@ -9,6 +9,7 @@ pub mod media;
 pub mod signalling;
 pub mod turn;
 
+use actix::actors::signal;
 use actix::prelude::*;
 use dotenv::dotenv;
 use log::prelude::*;
@@ -26,7 +27,6 @@ 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();
@@ -39,6 +39,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 +50,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,10 +1,13 @@
 //! Room definitions and implementations. Room is responsible for media
 //! connection establishment between concrete [`Member`]s.
 
+use core::fmt;
 use std::time::Duration;
 
+use actix::actors::signal;
+use actix::actors::signal::{ProcessSignals, Subscribe};
 use actix::{
-    fut::wrap_future, Actor, ActorFuture, AsyncContext, Context, Handler,
+    fut::wrap_future, Actor, ActorFuture, Addr, AsyncContext, Context, Handler,
     Message,
 };
 use failure::Fail;
@@ -37,7 +40,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 +92,6 @@ impl From<PeerStateError> for RoomError {
 }
 
 /// Media server room with its [`Member`]s.
-#[derive(Debug)]
 pub struct Room {
     id: Id,
 
@@ -74,6 +100,10 @@ pub struct Room {
 
     /// [`Peer`]s of [`Member`]s in this [`Room`].
     peers: PeerRepository,
+
+    // me attention
+    /// Actix addr of [`ProcessSignals`]
+    process_signals: Addr<ProcessSignals>,
 }
 
 impl Room {
@@ -84,6 +114,7 @@ impl Room {
         peers: HashMap<PeerId, PeerStateMachine>,
         reconnect_timeout: Duration,
         turn: Box<dyn TurnAuthService>,
+        process_signals: Addr<ProcessSignals>,
     ) -> Self {
         Self {
             id,
@@ -94,6 +125,7 @@ impl Room {
                 turn,
                 reconnect_timeout,
             ),
+            process_signals,
         }
     }
 
@@ -261,11 +293,28 @@ impl Room {
     }
 }
 
+impl fmt::Debug for Room {
+    // me attention
+    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) {
+        // me attention
+        self.process_signals
+            .do_send(Subscribe(ctx.address().recipient()));
+    }
 }
 
 impl Handler<Authorize> for Room {
@@ -429,6 +478,16 @@ impl Handler<CloseRoom> for Room {
     }
 }
 
+// Shutdown system on and of `SIGINT`, `SIGTERM`, `SIGQUIT` signals
+impl Handler<signal::Signal> for Room {
+    type Result = ();
+
+    // me attention
+    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 +533,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.