add server management commands

Cargo.toml

@@ -74,7 +74,7 @@ tokio-tls = { version="0.1", optional = true }
 tokio-openssl = { version="0.1", optional = true }
 
 [dependencies.actix]
-version = "^0.3.4"
+version = "^0.3.5"
 default-features = false
 features = []
 

src/lib.rs

@@ -71,8 +71,8 @@ extern crate brotli2;
 extern crate percent_encoding;
 extern crate smallvec;
 extern crate num_cpus;
-extern crate actix;
 extern crate h2 as http2;
+#[macro_use] extern crate actix;
 
 #[cfg(test)]
 #[macro_use] extern crate serde_derive;
@@ -173,7 +173,8 @@ pub mod dev {
     pub use pipeline::Pipeline;
     pub use channel::{HttpChannel, HttpHandler, IntoHttpHandler};
     pub use param::{FromParam, Params};
-    pub use server::ServerSettings;
     pub use httprequest::UrlEncoded;
     pub use httpresponse::HttpResponseBuilder;
+
+    pub use server::{ServerSettings, PauseServer, ResumeServer, StopServer};
 }

src/server.rs

@@ -1,7 +1,7 @@
 use std::{io, net, thread};
 use std::rc::Rc;
 use std::cell::{RefCell, RefMut};
-use std::sync::Arc;
+use std::sync::{Arc, mpsc as sync_mpsc};
 use std::time::Duration;
 use std::marker::PhantomData;
 use std::collections::HashMap;
@@ -106,6 +106,7 @@ pub struct HttpServer<T, A, H, U>
     factory: Arc<Fn() -> U + Send + Sync>,
     workers: Vec<SyncAddress<Worker<H>>>,
     sockets: HashMap<net::SocketAddr, net::TcpListener>,
+    accept: Vec<(mio::SetReadiness, sync_mpsc::Sender<Command>)>,
 }
 
 impl<T: 'static, A: 'static, H, U: 'static> Actor for HttpServer<T, A, H, U> {
@@ -144,6 +145,7 @@ impl<T, A, H, U, V> HttpServer<T, A, H, U>
                     factory: Arc::new(factory),
                     workers: Vec::new(),
                     sockets: HashMap::new(),
+                    accept: Vec::new(),
         }
     }
 
@@ -206,20 +208,11 @@ impl<T, A, H, U, V> HttpServer<T, A, H, U>
         let mut succ = false;
         if let Ok(iter) = addr.to_socket_addrs() {
             for addr in iter {
-                let builder = match addr {
+                match create_tcp_listener(addr, self.backlog) {
-                    net::SocketAddr::V4(_) => TcpBuilder::new_v4()?,
+                    Ok(lst) => {
-                    net::SocketAddr::V6(_) => TcpBuilder::new_v6()?,
-                };
-                match builder.bind(addr) {
-                    Ok(builder) => match builder.reuse_address(true) {
-                        Ok(builder) => {
                         succ = true;
-                            let lst = builder.listen(self.backlog)
-                                .expect("failed to set socket backlog");
                         self.sockets.insert(lst.local_addr().unwrap(), lst);
                     },
-                        Err(e) => err = Some(e)
-                    },
                     Err(e) => err = Some(e),
                 }
             }
@@ -309,7 +302,8 @@ impl<H: HttpHandler, U, V> HttpServer<TcpStream, net::SocketAddr, H, U>
             // start acceptors threads
             for (addr, sock) in addrs {
                 info!("Starting http server on {}", addr);
-                start_accept_thread(sock, addr, workers.clone());
+                self.accept.push(
+                    start_accept_thread(sock, addr, self.backlog, workers.clone()));
             }
 
             // start http server actor
@@ -328,7 +322,7 @@ impl<H: HttpHandler, U, V> HttpServer<TlsStream<TcpStream>, net::SocketAddr, H,
         if self.sockets.is_empty() {
             Err(io::Error::new(io::ErrorKind::Other, "No socket addresses are bound"))
         } else {
-            let addrs: Vec<(net::SocketAddr, Socket)> = self.sockets.drain().collect();
+            let addrs: Vec<(net::SocketAddr, net::TcpListener)> = self.sockets.drain().collect();
             let settings = ServerSettings::new(Some(addrs[0].0), &self.host, false);
             let acceptor = match TlsAcceptor::builder(pkcs12) {
                 Ok(builder) => {
@@ -344,7 +338,8 @@ impl<H: HttpHandler, U, V> HttpServer<TlsStream<TcpStream>, net::SocketAddr, H,
             // start acceptors threads
             for (addr, sock) in addrs {
                 info!("Starting tls http server on {}", addr);
-                start_accept_thread(sock, addr, workers.clone());
+                self.accept.push(
+                    start_accept_thread(sock, addr, self.backlog, workers.clone()));
             }
 
             // start http server actor
@@ -365,7 +360,7 @@ impl<H: HttpHandler, U, V> HttpServer<SslStream<TcpStream>, net::SocketAddr, H,
         if self.sockets.is_empty() {
             Err(io::Error::new(io::ErrorKind::Other, "No socket addresses are bound"))
         } else {
-            let addrs: Vec<(net::SocketAddr, Socket)> = self.sockets.drain().collect();
+            let addrs: Vec<(net::SocketAddr, net::TcpListener)> = self.sockets.drain().collect();
             let settings = ServerSettings::new(Some(addrs[0].0), &self.host, false);
             let acceptor = match SslAcceptorBuilder::mozilla_intermediate(
                 SslMethod::tls(), &identity.pkey, &identity.cert, &identity.chain)
@@ -383,7 +378,8 @@ impl<H: HttpHandler, U, V> HttpServer<SslStream<TcpStream>, net::SocketAddr, H,
             // start acceptors threads
             for (addr, sock) in addrs {
                 info!("Starting tls http server on {}", addr);
-                start_accept_thread(sock, addr, workers.clone());
+                self.accept.push(
+                    start_accept_thread(sock, addr, workers.clone(), self.backlog));
             }
 
             // start http server actor
@@ -414,7 +410,8 @@ impl<T, A, H, U, V> HttpServer<T, A, H, U>
             // start acceptors threads
             for (addr, sock) in addrs {
                 info!("Starting http server on {}", addr);
-                start_accept_thread(sock, addr, workers.clone());
+                self.accept.push(
+                    start_accept_thread(sock, addr, self.backlog, workers.clone()));
             }
         }
 
@@ -436,18 +433,13 @@ impl<T, A, H, U, V> HttpServer<T, A, H, U>
     }
 }
 
+#[derive(Message)]
 struct IoStream<T> {
     io: T,
     peer: Option<net::SocketAddr>,
     http2: bool,
 }
 
-impl<T> ResponseType for IoStream<T>
-{
-    type Item = ();
-    type Error = ();
-}
-
 impl<T, A, H, U> StreamHandler<IoStream<T>, io::Error> for HttpServer<T, A, H, U>
     where T: AsyncRead + AsyncWrite + 'static,
           H: HttpHandler + 'static,
@@ -473,6 +465,67 @@ impl<T, A, H, U> Handler<IoStream<T>, io::Error> for HttpServer<T, A, H, U>
     }
 }
 
+/// Pause connection accepting process
+#[derive(Message)]
+pub struct PauseServer;
+
+/// Resume connection accepting process
+#[derive(Message)]
+pub struct ResumeServer;
+
+/// Stop connection processing and exit
+#[derive(Message)]
+pub struct StopServer;
+
+impl<T, A, H, U> Handler<PauseServer> for HttpServer<T, A, H, U>
+    where T: AsyncRead + AsyncWrite + 'static,
+          H: HttpHandler + 'static,
+          U: 'static,
+          A: 'static,
+{
+    fn handle(&mut self, _: PauseServer, _: &mut Context<Self>) -> Response<Self, PauseServer>
+    {
+        for item in &self.accept {
+            let _ = item.1.send(Command::Pause);
+            let _ = item.0.set_readiness(mio::Ready::readable());
+        }
+        Self::empty()
+    }
+}
+
+impl<T, A, H, U> Handler<ResumeServer> for HttpServer<T, A, H, U>
+    where T: AsyncRead + AsyncWrite + 'static,
+          H: HttpHandler + 'static,
+          U: 'static,
+          A: 'static,
+{
+    fn handle(&mut self, _: ResumeServer, _: &mut Context<Self>) -> Response<Self, ResumeServer>
+    {
+        for item in &self.accept {
+            let _ = item.1.send(Command::Resume);
+            let _ = item.0.set_readiness(mio::Ready::readable());
+        }
+        Self::empty()
+    }
+}
+
+impl<T, A, H, U> Handler<StopServer> for HttpServer<T, A, H, U>
+    where T: AsyncRead + AsyncWrite + 'static,
+          H: HttpHandler + 'static,
+          U: 'static,
+          A: 'static,
+{
+    fn handle(&mut self, _: StopServer, ctx: &mut Context<Self>) -> Response<Self, StopServer>
+    {
+        for item in &self.accept {
+            let _ = item.1.send(Command::Stop);
+            let _ = item.0.set_readiness(mio::Ready::readable());
+        }
+        ctx.stop();
+        Self::empty()
+    }
+}
+
 /// Http worker
 ///
 /// Worker accepts Socket objects via unbounded channel and start requests processing.
@@ -589,10 +642,11 @@ impl StreamHandlerType {
                 let io = TcpStream::from_stream(io, hnd)
                     .expect("failed to associate TCP stream");
 
-                Arbiter::handle().spawn(
+                hnd.spawn(
                     TlsAcceptorExt::accept_async(acceptor, io).then(move |res| {
                         match res {
-                            Ok(io) => hnd.spawn(HttpChannel::new(h, io, peer, http2)),
+                            Ok(io) => Arbiter::handle().spawn(
+                                HttpChannel::new(h, io, peer, http2)),
                             Err(err) =>
                                 trace!("Error during handling tls connection: {}", err),
                         };
@@ -629,14 +683,27 @@ impl StreamHandlerType {
     }
 }
 
-fn start_accept_thread(sock: net::TcpListener, addr: net::SocketAddr,
+enum Command {
-                       workers: Vec<mpsc::UnboundedSender<IoStream<net::TcpStream>>>) {
+    Pause,
+    Resume,
+    Stop,
+}
+
+fn start_accept_thread(sock: net::TcpListener, addr: net::SocketAddr, backlog: i32,
+                       workers: Vec<mpsc::UnboundedSender<IoStream<net::TcpStream>>>)
+                       -> (mio::SetReadiness, sync_mpsc::Sender<Command>)
+{
+    let (tx, rx) = sync_mpsc::channel();
+    let (reg, readiness) = mio::Registration::new2();
+
     // start accept thread
     let _ = thread::Builder::new().name(format!("Accept on {}", addr)).spawn(move || {
-        let mut next = 0;
+        const SRV: mio::Token = mio::Token(0);
-        let server = mio::net::TcpListener::from_listener(sock, &addr)
+        const CMD: mio::Token = mio::Token(1);
-            .expect("Can not create mio::net::TcpListener");
+
-        const SERVER: mio::Token = mio::Token(0);
+        let mut server = Some(
+            mio::net::TcpListener::from_listener(sock, &addr)
+                .expect("Can not create mio::net::TcpListener"));
 
         // Create a poll instance
         let poll = match mio::Poll::new() {
@@ -645,14 +712,23 @@ fn start_accept_thread(sock: net::TcpListener, addr: net::SocketAddr,
         };
 
         // Start listening for incoming connections
-        if let Err(err) = poll.register(&server, SERVER,
+        if let Some(ref srv) = server {
-                                        mio::Ready::readable(), mio::PollOpt::edge()) {
+            if let Err(err) = poll.register(
+                srv, SRV, mio::Ready::readable(), mio::PollOpt::edge()) {
                 panic!("Can not register io: {}", err);
             }
+        }
+
+        // Start listening for incommin commands
+        if let Err(err) = poll.register(&reg, CMD,
+                                        mio::Ready::readable(), mio::PollOpt::edge()) {
+            panic!("Can not register Registration: {}", err);
+        }
 
         // Create storage for events
         let mut events = mio::Events::with_capacity(128);
 
+        let mut next = 0;
         loop {
             if let Err(err) = poll.poll(&mut events, None) {
                 panic!("Poll error: {}", err);
@@ -660,13 +736,15 @@ fn start_accept_thread(sock: net::TcpListener, addr: net::SocketAddr,
 
             for event in events.iter() {
                 match event.token() {
-                    SERVER => {
+                    SRV => {
+                        if let Some(ref server) = server {
                             loop {
                                 match server.accept_std() {
                                     Ok((sock, addr)) => {
-                                    let msg = IoStream{io: sock, peer: Some(addr), http2: false};
+                                        let msg = IoStream{
-                                    workers[next]
+                                            io: sock, peer: Some(addr), http2: false};
-                                        .unbounded_send(msg).expect("worker thread died");
+                                        workers[next].unbounded_send(msg)
+                                            .expect("worker thread died");
                                         next = (next + 1) % workers.len();
                                     },
                                     Err(err) => if err.kind() == io::ErrorKind::WouldBlock {
@@ -678,9 +756,57 @@ fn start_accept_thread(sock: net::TcpListener, addr: net::SocketAddr,
                                 }
                             }
                         }
+                    },
+                    CMD => match rx.try_recv() {
+                        Ok(cmd) => match cmd {
+                            Command::Pause => if let Some(server) = server.take() {
+                                if let Err(err) = poll.deregister(&server) {
+                                    error!("Can not deregister server socket {}", err);
+                                } else {
+                                    info!("Paused accepting connections on {}", addr);
+                                }
+                            },
+                            Command::Resume => {
+                                let lst = create_tcp_listener(addr, backlog)
+                                    .expect("Can not create net::TcpListener");
+
+                                server = Some(
+                                    mio::net::TcpListener::from_listener(lst, &addr)
+                                        .expect("Can not create mio::net::TcpListener"));
+
+                                if let Some(ref server) = server {
+                                    if let Err(err) = poll.register(
+                                        server, SRV, mio::Ready::readable(), mio::PollOpt::edge())
+                                    {
+                                        error!("Can not resume socket accept process: {}", err);
+                                    } else {
+                                        info!("Accepting connections on {} has been resumed",
+                                              addr);
+                                    }
+                                }
+                            },
+                            Command::Stop => return,
+                        },
+                        Err(err) => match err {
+                            sync_mpsc::TryRecvError::Empty => (),
+                            sync_mpsc::TryRecvError::Disconnected => return,
+                        }
+                    }
                     _ => unreachable!(),
                 }
             }
         }
     });
+
+    (readiness, tx)
+}
+
+fn create_tcp_listener(addr: net::SocketAddr, backlog: i32) -> io::Result<net::TcpListener> {
+    let builder = match addr {
+        net::SocketAddr::V4(_) => TcpBuilder::new_v4()?,
+        net::SocketAddr::V6(_) => TcpBuilder::new_v6()?,
+    };
+    builder.bind(addr)?;
+    builder.reuse_address(true)?;
+    Ok(builder.listen(backlog)?)
 }

tests/test_server.rs

@@ -2,10 +2,12 @@ extern crate actix;
 extern crate actix_web;
 extern crate tokio_core;
 extern crate reqwest;
+extern crate futures;
 
-use std::thread;
+use std::{net, thread};
 use std::sync::{Arc, mpsc};
 use std::sync::atomic::{AtomicUsize, Ordering};
+use futures::Future;
 
 use actix_web::*;
 use actix::System;
@@ -20,12 +22,22 @@ fn test_start() {
         let srv = HttpServer::new(
             || vec![Application::new()
                     .resource("/", |r| r.method(Method::GET).h(httpcodes::HTTPOk))]);
+
         let srv = srv.bind("127.0.0.1:0").unwrap();
-        let _ = tx.send(srv.addrs()[0].clone());
+        let addr = srv.addrs()[0].clone();
-        srv.start();
+        let srv_addr = srv.start();
+        let _ = tx.send((addr, srv_addr));
         sys.run();
     });
-    let addr = rx.recv().unwrap();
+    let (addr, srv_addr) = rx.recv().unwrap();
+    assert!(reqwest::get(&format!("http://{}/", addr)).unwrap().status().is_success());
+
+    // pause
+    let _ = srv_addr.call_fut(dev::PauseServer).wait();
+    assert!(net::TcpStream::connect(addr).is_err());
+
+    // resume
+    let _ = srv_addr.call_fut(dev::ResumeServer).wait();
     assert!(reqwest::get(&format!("http://{}/", addr)).unwrap().status().is_success());
 }