Graceful shutdown

Cargo.toml

@@ -21,6 +21,7 @@ lto = "thin"
 
 [dependencies]
 actix = "0.8"
+actix-rt = "0.2.3"
 actix-web = "1.0"
 actix-web-actors = "1.0"
 bb8 = "0.3"
@@ -49,7 +50,9 @@ slog-scope = "4.1"
 slog-stdlog = "3.0"
 smart-default = "0.5"
 tokio = "0.1"
+tokio-signal = "0.2.7"
 toml = "0.4"
+
 [dependencies.serde-humantime]
     git = "https://github.com/tailhook/serde-humantime"
     branch = "serde_wrapper"

config.toml

@@ -74,3 +74,9 @@
 #
 # Default:
 #   connection_timeout = "5s"
+
+
+[shutdown]
+# Number of milliseconds given to shutdown gracefully
+# Default:
+#   timeout = 5000

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;
@@ -27,7 +27,7 @@ pub trait RpcConnection: fmt::Debug + Send {
     /// Closes [`RpcConnection`].
     /// No [`RpcConnectionClosed`] signals should be emitted.
     /// Always returns success.
-    fn close(&mut self) -> Box<dyn Future<Item = (), Error = ()>>;
+    fn close(&mut self) -> Box<dyn Future<Item = (), Error = ()> + Send>;
 
     /// Sends [`Event`] to remote [`Member`].
     fn send_event(
@@ -208,7 +208,7 @@ pub mod test {
     }
 
     impl RpcConnection for Addr<TestConnection> {
-        fn close(&mut self) -> Box<dyn Future<Item = (), Error = ()>> {
+        fn close(&mut self) -> Box<dyn Future<Item = (), Error = ()> + Send> {
             let fut = self.send(Close {}).map_err(|_| ());
             Box::new(fut)
         }

src/api/client/server.rs

@@ -1,9 +1,9 @@
 //! HTTP server for handling WebSocket connections of Client API.
 
+use actix::{Actor, Addr};
 use actix_web::{
-    middleware,
-    web::{resource, Data, Path, Payload},
     App, HttpRequest, HttpResponse, HttpServer,
+    middleware, web::{Data, Path, Payload, resource},
 };
 use actix_web_actors::ws;
 use futures::{
@@ -25,6 +25,8 @@ use crate::{
     signalling::{RoomId, RoomsRepository},
 };
 
+use self::actors::ServerWrapper;
+
 /// Parameters of new WebSocket connection creation HTTP request.
 #[derive(Debug, Deserialize)]
 struct RequestParams {
@@ -85,10 +87,10 @@ pub struct Context {
 }
 
 /// Starts HTTP server for handling WebSocket connections of Client API.
-pub fn run(rooms: RoomsRepository, config: Conf) {
+pub fn run(rooms: RoomsRepository, config: Conf) -> Addr<ServerWrapper> {
     let server_addr = config.server.bind_addr();
 
-    HttpServer::new(move || {
+    let actix_server = HttpServer::new(move || {
         App::new()
             .data(Context {
                 rooms: rooms.clone(),
@@ -100,19 +102,52 @@ pub fn run(rooms: RoomsRepository, config: Conf) {
                     .route(actix_web::web::get().to_async(ws_index)),
             )
     })
+    .disable_signals()
     .bind(server_addr)
     .unwrap()
     .start();
 
-    info!("Started HTTP server on 0.0.0.0:8080");
+    info!("Started HTTP server on {:?}", server_addr);
+
+    actors::ServerWrapper(actix_server).start()
+}
+
+pub mod actors {
+    use actix::{Actor, ActorFuture, Context, Handler};
+    use actix::fut::wrap_future;
+    use actix_web::dev::Server;
+    use tokio::prelude::Future;
+
+    use crate::{log::prelude::*, shutdown::ShutdownMessage};
+
+    pub struct ServerWrapper(pub Server);
+
+    impl Actor for ServerWrapper {
+        type Context = Context<Self>;
+    }
+
+    impl Handler<ShutdownMessage> for ServerWrapper {
+        type Result = Box<dyn ActorFuture<Actor = Self, Item = (), Error = ()>>;
+
+        fn handle(
+            &mut self,
+            _: ShutdownMessage,
+            _: &mut Self::Context,
+        ) -> Self::Result {
+            info!("Shutting down Actix Web Server");
+
+            Box::new(wrap_future(
+                self.0.stop(true).then(move |_| futures::future::ok(())),
+            ))
+        }
+    }
 }
 
 #[cfg(test)]
 mod test {
     use std::{ops::Add, thread, time::Duration};
 
-    use actix::Actor as _;
-    use actix_http::{ws::Message, HttpService};
+    use actix_http::{HttpService, ws::Message};
     use actix_http_test::{TestServer, TestServerRuntime};
     use futures::{future::IntoFuture as _, sink::Sink as _, Stream as _};
 
@@ -171,13 +206,15 @@ mod test {
 
     #[test]
     fn ping_pong_and_disconnects_on_idle() {
+        let config = Conf::parse().unwrap();
         let conf = Conf {
             rpc: Rpc {
                 idle_timeout: Duration::new(2, 0),
                 reconnect_timeout: Default::default(),
             },
             turn: Turn::default(),
             server: Server::default(),
+            system_config: config.system_config,
         };
 
         let mut server = ws_server(conf.clone());

src/api/client/session.rs

@@ -149,7 +149,7 @@ impl RpcConnection for Addr<WsSession> {
     /// Closes [`WsSession`] by sending itself "normal closure" close message.
     ///
     /// Never returns error.
-    fn close(&mut self) -> Box<dyn Future<Item = (), Error = ()>> {
+    fn close(&mut self) -> Box<dyn Future<Item = (), Error = ()> + Send> {
         let fut = self
             .send(Close {
                 reason: Some(ws::CloseCode::Normal.into()),

src/conf/mod.rs

@@ -2,6 +2,7 @@
 
 pub mod rpc;
 pub mod server;
+pub mod shutdown_config;
 pub mod turn;
 
 use std::env;
@@ -13,6 +14,7 @@ use serde::{Deserialize, Serialize};
 pub use self::{
     rpc::Rpc,
     server::Server,
+    shutdown_config::ShutdownConfiguration,
     turn::{Redis, Turn},
 };
 
@@ -33,6 +35,8 @@ pub struct Conf {
     pub server: Server,
     /// TURN server settings.
     pub turn: Turn,
+    // More settings
+    pub system_config: ShutdownConfiguration,
 }
 
 impl Conf {

src/conf/shutdown_config.rs

@@ -0,0 +1,11 @@
+//! More system settings
+
+use serde::{Deserialize, Serialize};
+use smart_default::*;
+
+#[derive(Clone, Debug, Deserialize, Serialize, SmartDefault)]
+#[serde(default)]
+pub struct ShutdownConfiguration {
+    #[default(5000)]
+    pub timeout: u64,
+}

src/main.rs

@@ -1,15 +1,7 @@
 //! Medea media server application.
 
-#[macro_use]
-pub mod utils;
-pub mod api;
-pub mod conf;
-pub mod log;
-pub mod media;
-pub mod signalling;
-pub mod turn;
-
 use actix::prelude::*;
+
 use dotenv::dotenv;
 use log::prelude::*;
 
@@ -21,24 +13,36 @@ use crate::{
     turn::new_turn_auth_service,
 };
 
+#[macro_use]
+pub mod utils;
+pub mod api;
+pub mod conf;
+pub mod log;
+pub mod media;
+pub mod shutdown;
+pub mod signalling;
+pub mod turn;
+
 fn main() {
     dotenv().ok();
     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);
 
+    let sys = System::new("medea");
+
     let members = hashmap! {
         1 => Member::new(1, "caller_credentials".to_owned()),
         2 => Member::new(2, "responder_credentials".to_owned()),
     };
+
     let peers = create_peers(1, 2);
 
+    let graceful_shutdown_addr = shutdown::create(config.system_config.timeout);
+
     let turn_auth_service =
         new_turn_auth_service(&config).expect("Unable to start turn service");
 
@@ -47,10 +51,19 @@ fn main() {
     let room = Room::start_in_arbiter(&Arbiter::new(), move |_| {
         Room::new(1, members, peers, rpc_reconnect_timeout, turn_auth_service)
     });
+    graceful_shutdown_addr.do_send(shutdown::Subscribe(shutdown::Subscriber {
+        addr: room.clone().recipient(),
+        priority: shutdown::Priority(1),
+    }));
 
     let rooms = hashmap! {1 => room};
     let rooms_repo = RoomsRepository::new(rooms);
 
-    server::run(rooms_repo, config);
+    let server_addr = server::run(rooms_repo, config);
+    graceful_shutdown_addr.do_send(shutdown::Subscribe(shutdown::Subscriber {
+        addr: server_addr.recipient(),
+        priority: shutdown::Priority(5),
+    }));
+
     let _ = sys.run();
 }