add StopWorker message

src/server.rs

@@ -15,28 +15,24 @@ use mio;
 use num_cpus;
 use net2::TcpBuilder;
 
-#[cfg(feature="tls")]
-use futures::{future, Future};
 #[cfg(feature="tls")]
 use native_tls::TlsAcceptor;
 #[cfg(feature="tls")]
-use tokio_tls::{TlsStream, TlsAcceptorExt};
+use tokio_tls::TlsStream;
 
 #[cfg(feature="alpn")]
-use futures::{future, Future};
+use openssl::ssl::{SslMethod, SslAcceptorBuilder};
-#[cfg(feature="alpn")]
-use openssl::ssl::{SslMethod, SslAcceptor, SslAcceptorBuilder};
 #[cfg(feature="alpn")]
 use openssl::pkcs12::ParsedPkcs12;
 #[cfg(feature="alpn")]
-use tokio_openssl::{SslStream, SslAcceptorExt};
+use tokio_openssl::SslStream;
 
 #[cfg(feature="signal")]
 use actix::actors::signal;
 
 use helpers;
 use channel::{HttpChannel, HttpHandler, IntoHttpHandler};
-use worker::{Conn, Worker, WorkerSettings, StreamHandlerType};
+use worker::{Conn, Worker, WorkerSettings, StreamHandlerType, StopWorker};
 
 /// Various server settings
 #[derive(Debug, Clone)]
@@ -604,14 +600,21 @@ impl<T, A, H, U> Handler<StopServer> for HttpServer<T, A, H, U>
           U: 'static,
           A: 'static,
 {
-    fn handle(&mut self, _: StopServer, ctx: &mut Context<Self>) -> Response<Self, StopServer>
+    fn handle(&mut self, msg: StopServer, ctx: &mut Context<Self>) -> Response<Self, StopServer>
     {
+        // stop accept threads
         for item in &self.accept {
             let _ = item.1.send(Command::Stop);
             let _ = item.0.set_readiness(mio::Ready::readable());
         }
         ctx.stop();
 
+        // stop workers
+        let dur = if msg.graceful { Some(Duration::new(30, 0)) } else { None };
+        for worker in &self.workers {
+            worker.send(StopWorker{graceful: dur})
+        }
+
         // we need to stop system if server was spawned
         if self.exit {
             Arbiter::system().send(msgs::SystemExit(0))

src/worker.rs

@@ -5,7 +5,22 @@ use tokio_core::net::TcpStream;
 use tokio_core::reactor::Handle;
 use net2::TcpStreamExt;
 
+#[cfg(feature="tls")]
+use futures::{future, Future};
+#[cfg(feature="tls")]
+use native_tls::TlsAcceptor;
+#[cfg(feature="tls")]
+use tokio_tls::TlsAcceptorExt;
+
+#[cfg(feature="alpn")]
+use futures::{future, Future};
+#[cfg(feature="alpn")]
+use openssl::ssl::SslAcceptor;
+#[cfg(feature="alpn")]
+use tokio_openssl::SslAcceptorExt;
+
 use actix::{Actor, Arbiter, AsyncContext, Context, Handler, Response, StreamHandler};
+use actix::msgs::StopArbiter;
 
 use helpers;
 use channel::{HttpChannel, HttpHandler};
@@ -18,6 +33,12 @@ pub(crate) struct Conn<T> {
     pub http2: bool,
 }
 
+/// Stop worker
+#[derive(Message)]
+pub(crate) struct StopWorker {
+    pub graceful: Option<time::Duration>,
+}
+
 pub(crate) struct WorkerSettings<H> {
     h: RefCell<Vec<H>>,
     enabled: bool,
@@ -108,6 +129,17 @@ impl<H> Handler<Conn<net::TcpStream>> for Worker<H>
     }
 }
 
+/// `StopWorker` message handler
+impl<H> Handler<StopWorker> for Worker<H>
+    where H: HttpHandler + 'static,
+{
+    fn handle(&mut self, _: StopWorker, _: &mut Context<Self>) -> Response<Self, StopWorker>
+    {
+        Arbiter::arbiter().send(StopArbiter(0));
+        Self::empty()
+    }
+}
+
 #[derive(Clone)]
 pub(crate) enum StreamHandlerType {
     Normal,