fix node::insert() method, missing next element

Fix issue with HttpChannel linked list.

CHANGES.md

@@ -1,5 +1,42 @@
 # Changes
 
+## [0.7.8] - 2018-09-17
+
+### Added
+
+* Use server `Keep-Alive` setting as slow request timeout #439
+
+### Changed
+
+* Use 5 seconds keep-alive timer by default.
+
+### Fixed
+
+* Fixed wrong error message for i16 type #510
+
+
+## [0.7.7] - 2018-09-11
+
+### Fixed
+
+* Fix linked list of HttpChannels #504
+
+* Fix requests to TestServer fail #508
+
+
+## [0.7.6] - 2018-09-07
+
+### Fixed
+
+* Fix system_exit in HttpServer #501
+
+* Fix parsing of route param containin regexes with repetition #500
+
+### Changes
+
+* Unhide `SessionBackend` and `SessionImpl` traits #455
+
+
 ## [0.7.5] - 2018-09-04
 
 ### Added

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-web"
-version = "0.7.5"
+version = "0.7.8"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix web is a simple, pragmatic and extremely fast web framework for Rust."
 readme = "README.md"
@@ -101,6 +101,7 @@ tokio-io = "0.1"
 tokio-tcp = "0.1"
 tokio-timer = "0.2"
 tokio-reactor = "0.1"
+tokio-current-thread = "0.1"
 
 # native-tls
 native-tls = { version="0.2", optional = true }

src/client/connector.rs

@@ -204,6 +204,7 @@ impl Paused {
 /// `ClientConnector` type is responsible for transport layer of a
 /// client connection.
 pub struct ClientConnector {
+    #[allow(dead_code)]
     connector: SslConnector,
 
     stats: ClientConnectorStats,

src/client/request.rs

@@ -254,16 +254,16 @@ impl ClientRequest {
 
 impl fmt::Debug for ClientRequest {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        let res = writeln!(
+        writeln!(
             f,
             "\nClientRequest {:?} {}:{}",
             self.version, self.method, self.uri
-        );
+        )?;
-        let _ = writeln!(f, "  headers:");
+        writeln!(f, "  headers:")?;
         for (key, val) in self.headers.iter() {
-            let _ = writeln!(f, "    {:?}: {:?}", key, val);
+            writeln!(f, "    {:?}: {:?}", key, val)?;
         }
-        res
+        Ok(())
     }
 }
 
@@ -750,16 +750,16 @@ fn parts<'a>(
 impl fmt::Debug for ClientRequestBuilder {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         if let Some(ref parts) = self.request {
-            let res = writeln!(
+            writeln!(
                 f,
                 "\nClientRequestBuilder {:?} {}:{}",
                 parts.version, parts.method, parts.uri
-            );
+            )?;
-            let _ = writeln!(f, "  headers:");
+            writeln!(f, "  headers:")?;
             for (key, val) in parts.headers.iter() {
-                let _ = writeln!(f, "    {:?}: {:?}", key, val);
+                writeln!(f, "    {:?}: {:?}", key, val)?;
             }
-            res
+            Ok(())
         } else {
             write!(f, "ClientRequestBuilder(Consumed)")
         }

src/client/response.rs

@@ -95,12 +95,12 @@ impl ClientResponse {
 
 impl fmt::Debug for ClientResponse {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        let res = writeln!(f, "\nClientResponse {:?} {}", self.version(), self.status());
+        writeln!(f, "\nClientResponse {:?} {}", self.version(), self.status())?;
-        let _ = writeln!(f, "  headers:");
+        writeln!(f, "  headers:")?;
         for (key, val) in self.headers().iter() {
-            let _ = writeln!(f, "    {:?}: {:?}", key, val);
+            writeln!(f, "    {:?}: {:?}", key, val)?;
         }
-        res
+        Ok(())
     }
 }
 

src/de.rs

@@ -175,7 +175,7 @@ impl<'de, S: 'de> Deserializer<'de> for PathDeserializer<'de, S> {
     parse_single_value!(deserialize_bool, visit_bool, "bool");
     parse_single_value!(deserialize_i8, visit_i8, "i8");
     parse_single_value!(deserialize_i16, visit_i16, "i16");
-    parse_single_value!(deserialize_i32, visit_i32, "i16");
+    parse_single_value!(deserialize_i32, visit_i32, "i32");
     parse_single_value!(deserialize_i64, visit_i64, "i64");
     parse_single_value!(deserialize_u8, visit_u8, "u8");
     parse_single_value!(deserialize_u16, visit_u16, "u16");

src/httprequest.rs

@@ -354,24 +354,24 @@ impl<S> FromRequest<S> for HttpRequest<S> {
 
 impl<S> fmt::Debug for HttpRequest<S> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        let res = writeln!(
+        writeln!(
             f,
             "\nHttpRequest {:?} {}:{}",
             self.version(),
             self.method(),
             self.path()
-        );
+        )?;
         if !self.query_string().is_empty() {
-            let _ = writeln!(f, "  query: ?{:?}", self.query_string());
+            writeln!(f, "  query: ?{:?}", self.query_string())?;
         }
         if !self.match_info().is_empty() {
-            let _ = writeln!(f, "  params: {:?}", self.match_info());
+            writeln!(f, "  params: {:?}", self.match_info())?;
         }
-        let _ = writeln!(f, "  headers:");
+        writeln!(f, "  headers:")?;
         for (key, val) in self.headers().iter() {
-            let _ = writeln!(f, "    {:?}: {:?}", key, val);
+            writeln!(f, "    {:?}: {:?}", key, val)?;
         }
-        res
+        Ok(())
     }
 }
 

src/lib.rs

@@ -118,6 +118,7 @@ extern crate parking_lot;
 extern crate rand;
 extern crate slab;
 extern crate tokio;
+extern crate tokio_current_thread;
 extern crate tokio_io;
 extern crate tokio_reactor;
 extern crate tokio_tcp;

src/middleware/session.rs

@@ -270,14 +270,17 @@ impl<S: 'static, T: SessionBackend<S>> Middleware<S> for SessionStorage<T, S> {
 }
 
 /// A simple key-value storage interface that is internally used by `Session`.
-#[doc(hidden)]
 pub trait SessionImpl: 'static {
+    /// Get session value by key
     fn get(&self, key: &str) -> Option<&str>;
 
+    /// Set session value
     fn set(&mut self, key: &str, value: String);
 
+    /// Remove specific key from session
     fn remove(&mut self, key: &str);
 
+    /// Remove all values from session
     fn clear(&mut self);
 
     /// Write session to storage backend.
@@ -285,9 +288,10 @@ pub trait SessionImpl: 'static {
 }
 
 /// Session's storage backend trait definition.
-#[doc(hidden)]
 pub trait SessionBackend<S>: Sized + 'static {
+    /// Session item
     type Session: SessionImpl;
+    /// Future that reads session
     type ReadFuture: Future<Item = Self::Session, Error = Error>;
 
     /// Parse the session from request and load data from a storage backend.

src/multipart.rs

@@ -441,13 +441,13 @@ where
 
 impl<S> fmt::Debug for Field<S> {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        let res = writeln!(f, "\nMultipartField: {}", self.ct);
+        writeln!(f, "\nMultipartField: {}", self.ct)?;
-        let _ = writeln!(f, "  boundary: {}", self.inner.borrow().boundary);
+        writeln!(f, "  boundary: {}", self.inner.borrow().boundary)?;
-        let _ = writeln!(f, "  headers:");
+        writeln!(f, "  headers:")?;
         for (key, val) in self.headers.iter() {
-            let _ = writeln!(f, "    {:?}: {:?}", key, val);
+            writeln!(f, "    {:?}: {:?}", key, val)?;
         }
-        res
+        Ok(())
     }
 }
 

src/router.rs

@@ -815,73 +815,70 @@ impl ResourceDef {
         Ok(())
     }
 
-    fn parse(
+    fn parse_param(pattern: &str) -> (PatternElement, String, &str) {
-        pattern: &str, for_prefix: bool,
-    ) -> (String, Vec<PatternElement>, bool, usize) {
         const DEFAULT_PATTERN: &str = "[^/]+";
-
+        let mut params_nesting = 0usize;
-        let mut re1 = String::from("^");
+        let close_idx = pattern
-        let mut re2 = String::new();
+            .find(|c| match c {
-        let mut el = String::new();
+                '{' => {
-        let mut in_param = false;
+                    params_nesting += 1;
-        let mut in_param_pattern = false;
+                    false
-        let mut param_name = String::new();
-        let mut param_pattern = String::from(DEFAULT_PATTERN);
-        let mut is_dynamic = false;
-        let mut elems = Vec::new();
-        let mut len = 0;
-
-        for ch in pattern.chars() {
-            if in_param {
-                // In parameter segment: `{....}`
-                if ch == '}' {
-                    elems.push(PatternElement::Var(param_name.clone()));
-                    re1.push_str(&format!(r"(?P<{}>{})", &param_name, &param_pattern));
-
-                    param_name.clear();
-                    param_pattern = String::from(DEFAULT_PATTERN);
-
-                    len = 0;
-                    in_param_pattern = false;
-                    in_param = false;
-                } else if ch == ':' {
-                    // The parameter name has been determined; custom pattern land
-                    in_param_pattern = true;
-                    param_pattern.clear();
-                } else if in_param_pattern {
-                    // Ignore leading whitespace for pattern
-                    if !(ch == ' ' && param_pattern.is_empty()) {
-                        param_pattern.push(ch);
-                    }
-                } else {
-                    param_name.push(ch);
                 }
-            } else if ch == '{' {
+                '}' => {
-                in_param = true;
+                    params_nesting -= 1;
-                is_dynamic = true;
+                    params_nesting == 0
-                elems.push(PatternElement::Str(el.clone()));
+                }
-                el.clear();
+                _ => false,
-            } else {
+            }).expect("malformed param");
-                re1.push_str(escape(&ch.to_string()).as_str());
+        let (mut param, rem) = pattern.split_at(close_idx + 1);
-                re2.push(ch);
+        param = &param[1..param.len() - 1]; // Remove outer brackets
-                el.push(ch);
+        let (name, pattern) = match param.find(':') {
-                len += 1;
+            Some(idx) => {
+                let (name, pattern) = param.split_at(idx);
+                (name, &pattern[1..])
             }
-        }
+            None => (param, DEFAULT_PATTERN),
-
-        if !el.is_empty() {
-            elems.push(PatternElement::Str(el.clone()));
-        }
-
-        let re = if is_dynamic {
-            if !for_prefix {
-                re1.push('$');
-            }
-            re1
-        } else {
-            re2
         };
-        (re, elems, is_dynamic, len)
+        (
+            PatternElement::Var(name.to_string()),
+            format!(r"(?P<{}>{})", &name, &pattern),
+            rem,
+        )
+    }
+
+    fn parse(
+        mut pattern: &str, for_prefix: bool,
+    ) -> (String, Vec<PatternElement>, bool, usize) {
+        if pattern.find('{').is_none() {
+            return (
+                String::from(pattern),
+                vec![PatternElement::Str(String::from(pattern))],
+                false,
+                pattern.chars().count(),
+            );
+        };
+
+        let mut elems = Vec::new();
+        let mut re = String::from("^");
+
+        while let Some(idx) = pattern.find('{') {
+            let (prefix, rem) = pattern.split_at(idx);
+            elems.push(PatternElement::Str(String::from(prefix)));
+            re.push_str(&escape(prefix));
+            let (param_pattern, re_part, rem) = Self::parse_param(rem);
+            elems.push(param_pattern);
+            re.push_str(&re_part);
+            pattern = rem;
+        }
+
+        elems.push(PatternElement::Str(String::from(pattern)));
+        re.push_str(&escape(pattern));
+
+        if !for_prefix {
+            re.push_str("$");
+        }
+
+        (re, elems, true, pattern.chars().count())
     }
 }
 
@@ -1072,6 +1069,16 @@ mod tests {
         let info = re.match_with_params(&req, 0).unwrap();
         assert_eq!(info.get("version").unwrap(), "151");
         assert_eq!(info.get("id").unwrap(), "adahg32");
+
+        let re = ResourceDef::new("/{id:[[:digit:]]{6}}");
+        assert!(re.is_match("/012345"));
+        assert!(!re.is_match("/012"));
+        assert!(!re.is_match("/01234567"));
+        assert!(!re.is_match("/XXXXXX"));
+
+        let req = TestRequest::with_uri("/012345").finish();
+        let info = re.match_with_params(&req, 0).unwrap();
+        assert_eq!(info.get("id").unwrap(), "012345");
     }
 
     #[test]

src/scope.rs

@@ -821,11 +821,9 @@ mod tests {
                 scope
                     .route("/path1", Method::GET, |_: HttpRequest<_>| {
                         HttpResponse::Ok()
-                    }).route(
+                    }).route("/path1", Method::DELETE, |_: HttpRequest<_>| {
-                        "/path1",
+                        HttpResponse::Ok()
-                        Method::DELETE,
+                    })
-                        |_: HttpRequest<_>| HttpResponse::Ok(),
-                    )
             }).finish();
 
         let req = TestRequest::with_uri("/app/path1").request();

src/server/accept.rs

@@ -451,10 +451,13 @@ impl Accept {
                                     Delay::new(
                                         Instant::now() + Duration::from_millis(510),
                                     ).map_err(|_| ())
-                                    .and_then(move |_| {
+                                    .and_then(
-                                        let _ = r.set_readiness(mio::Ready::readable());
+                                        move |_| {
-                                        Ok(())
+                                            let _ =
-                                    }),
+                                                r.set_readiness(mio::Ready::readable());
+                                            Ok(())
+                                        },
+                                    ),
                                 );
                                 Ok(())
                             },

src/server/channel.rs

@@ -5,6 +5,7 @@ use std::{io, ptr, time};
 use bytes::{Buf, BufMut, BytesMut};
 use futures::{Async, Future, Poll};
 use tokio_io::{AsyncRead, AsyncWrite};
+use tokio_timer::Delay;
 
 use super::settings::WorkerSettings;
 use super::{h1, h2, ConnectionTag, HttpHandler, IoStream};
@@ -30,6 +31,7 @@ where
 {
     proto: Option<HttpProtocol<T, H>>,
     node: Option<Node<HttpChannel<T, H>>>,
+    ka_timeout: Option<Delay>,
     _tag: ConnectionTag,
 }
 
@@ -42,9 +44,11 @@ where
         settings: Rc<WorkerSettings<H>>, io: T, peer: Option<SocketAddr>,
     ) -> HttpChannel<T, H> {
         let _tag = settings.connection();
+        let ka_timeout = settings.keep_alive_timer();
 
         HttpChannel {
             _tag,
+            ka_timeout,
             node: None,
             proto: Some(HttpProtocol::Unknown(
                 settings,
@@ -55,7 +59,7 @@ where
         }
     }
 
-    fn shutdown(&mut self) {
+    pub(crate) fn shutdown(&mut self) {
         match self.proto {
             Some(HttpProtocol::H1(ref mut h1)) => {
                 let io = h1.io();
@@ -68,6 +72,18 @@ where
     }
 }
 
+impl<T, H> Drop for HttpChannel<T, H>
+where
+    T: IoStream,
+    H: HttpHandler + 'static,
+{
+    fn drop(&mut self) {
+        if let Some(mut node) = self.node.take() {
+            node.remove()
+        }
+    }
+}
+
 impl<T, H> Future for HttpChannel<T, H>
 where
     T: IoStream,
@@ -77,7 +93,19 @@ where
     type Error = ();
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        if self.node.is_some() {
+        // keep-alive timer
+        if let Some(ref mut timer) = self.ka_timeout {
+            match timer.poll() {
+                Ok(Async::Ready(_)) => {
+                    trace!("Slow request timed out, close connection");
+                    return Ok(Async::Ready(()));
+                }
+                Ok(Async::NotReady) => (),
+                Err(_) => panic!("Something is really wrong"),
+            }
+        }
+
+        if self.node.is_none() {
             let el = self as *mut _;
             self.node = Some(Node::new(el));
             let _ = match self.proto {
@@ -97,28 +125,10 @@ where
         let mut is_eof = false;
         let kind = match self.proto {
             Some(HttpProtocol::H1(ref mut h1)) => {
-                let result = h1.poll();
+                return h1.poll();
-                match result {
-                    Ok(Async::Ready(())) | Err(_) => {
-                        if let Some(n) = self.node.as_mut() {
-                            n.remove()
-                        };
-                    }
-                    _ => (),
-                }
-                return result;
             }
             Some(HttpProtocol::H2(ref mut h2)) => {
-                let result = h2.poll();
+                return h2.poll();
-                match result {
-                    Ok(Async::Ready(())) | Err(_) => {
-                        if let Some(n) = self.node.as_mut() {
-                            n.remove()
-                        };
-                    }
-                    _ => (),
-                }
-                return result;
             }
             Some(HttpProtocol::Unknown(_, _, ref mut io, ref mut buf)) => {
                 let mut disconnect = false;
@@ -137,9 +147,6 @@ where
                 }
                 if disconnect {
                     debug!("Ignored premature client disconnection");
-                    if let Some(n) = self.node.as_mut() {
-                        n.remove()
-                    };
                     return Err(());
                 }
 
@@ -161,7 +168,12 @@ where
             match kind {
                 ProtocolKind::Http1 => {
                     self.proto = Some(HttpProtocol::H1(h1::Http1::new(
-                        settings, io, addr, buf, is_eof,
+                        settings,
+                        io,
+                        addr,
+                        buf,
+                        is_eof,
+                        self.ka_timeout.take(),
                     )));
                     return self.poll();
                 }
@@ -171,6 +183,7 @@ where
                         io,
                         addr,
                         buf.freeze(),
+                        self.ka_timeout.take(),
                     )));
                     return self.poll();
                 }
@@ -195,13 +208,14 @@ impl<T> Node<T> {
         }
     }
 
-    fn insert<I>(&mut self, next: &mut Node<I>) {
+    fn insert<I>(&mut self, next_el: &mut Node<I>) {
         unsafe {
-            let next: *mut Node<T> = next as *const _ as *mut _;
+            let next: *mut Node<T> = next_el as *const _ as *mut _;
 
-            if let Some(ref mut next2) = self.next {
+            if let Some(next2) = self.next {
                 let n = next2.as_mut().unwrap();
                 n.prev = Some(next);
+                next_el.next = Some(next2 as *mut _);
             }
             self.next = Some(next);
 
@@ -214,11 +228,14 @@ impl<T> Node<T> {
         unsafe {
             self.element = ptr::null_mut();
             let next = self.next.take();
-            let mut prev = self.prev.take();
+            let prev = self.prev.take();
 
-            if let Some(ref mut prev) = prev {
+            if let Some(prev) = prev {
                 prev.as_mut().unwrap().next = next;
             }
+            if let Some(next) = next {
+                next.as_mut().unwrap().prev = prev;
+            }
         }
     }
 }
@@ -232,7 +249,7 @@ impl Node<()> {
         }
     }
 
-    pub(crate) fn traverse<T, H>(&self)
+    pub(crate) fn traverse<T, H, F: Fn(&mut HttpChannel<T, H>)>(&self, f: F)
     where
         T: IoStream,
         H: HttpHandler + 'static,
@@ -247,7 +264,7 @@ impl Node<()> {
                     if !n.element.is_null() {
                         let ch: &mut HttpChannel<T, H> =
                             &mut *(&mut *(n.element as *mut _) as *mut () as *mut _);
-                        ch.shutdown();
+                        f(ch);
                     }
                 }
             } else {

src/server/h1.rs

@@ -91,7 +91,7 @@ where
 {
     pub fn new(
         settings: Rc<WorkerSettings<H>>, stream: T, addr: Option<SocketAddr>,
-        buf: BytesMut, is_eof: bool,
+        buf: BytesMut, is_eof: bool, keepalive_timer: Option<Delay>,
     ) -> Self {
         Http1 {
             flags: if is_eof {
@@ -103,10 +103,10 @@ where
             decoder: H1Decoder::new(),
             payload: None,
             tasks: VecDeque::new(),
-            keepalive_timer: None,
             addr,
             buf,
             settings,
+            keepalive_timer,
         }
     }
 
@@ -364,7 +364,7 @@ where
             if self.keepalive_timer.is_none() && keep_alive > 0 {
                 trace!("Start keep-alive timer");
                 let mut timer =
-                    Delay::new(Instant::now() + Duration::new(keep_alive, 0));
+                    Delay::new(Instant::now() + Duration::from_secs(keep_alive));
                 // register timer
                 let _ = timer.poll();
                 self.keepalive_timer = Some(timer);
@@ -632,7 +632,7 @@ mod tests {
         let readbuf = BytesMut::new();
         let settings = Rc::new(wrk_settings());
 
-        let mut h1 = Http1::new(Rc::clone(&settings), buf, None, readbuf, false);
+        let mut h1 = Http1::new(Rc::clone(&settings), buf, None, readbuf, false, None);
         h1.poll_io();
         h1.poll_io();
         assert_eq!(h1.tasks.len(), 1);
@@ -645,7 +645,7 @@ mod tests {
             BytesMut::from(Vec::<u8>::from(&b"GET /test HTTP/1.1\r\n\r\n"[..]));
         let settings = Rc::new(wrk_settings());
 
-        let mut h1 = Http1::new(Rc::clone(&settings), buf, None, readbuf, true);
+        let mut h1 = Http1::new(Rc::clone(&settings), buf, None, readbuf, true, None);
         h1.poll_io();
         assert_eq!(h1.tasks.len(), 1);
     }
@@ -656,7 +656,7 @@ mod tests {
         let readbuf = BytesMut::new();
         let settings = Rc::new(wrk_settings());
 
-        let mut h1 = Http1::new(Rc::clone(&settings), buf, None, readbuf, false);
+        let mut h1 = Http1::new(Rc::clone(&settings), buf, None, readbuf, false, None);
         h1.poll_io();
         h1.poll_io();
         assert!(h1.flags.contains(Flags::ERROR));

src/server/h2.rs

@@ -59,6 +59,7 @@ where
 {
     pub fn new(
         settings: Rc<WorkerSettings<H>>, io: T, addr: Option<SocketAddr>, buf: Bytes,
+        keepalive_timer: Option<Delay>,
     ) -> Self {
         let extensions = io.extensions();
         Http2 {
@@ -68,10 +69,10 @@ where
                 unread: if buf.is_empty() { None } else { Some(buf) },
                 inner: io,
             })),
-            keepalive_timer: None,
             addr,
             settings,
             extensions,
+            keepalive_timer,
         }
     }
 

src/server/http.rs

@@ -8,7 +8,6 @@ use actix::{Actor, Addr, Arbiter, AsyncContext, Context, Handler, System};
 use futures::{Future, Stream};
 use net2::{TcpBuilder, TcpStreamExt};
 use num_cpus;
-use tokio::executor::current_thread;
 use tokio_io::{AsyncRead, AsyncWrite};
 use tokio_tcp::TcpStream;
 
@@ -71,9 +70,9 @@ where
             factory: Arc::new(f),
             host: None,
             backlog: 2048,
-            keep_alive: KeepAlive::Os,
+            keep_alive: KeepAlive::Timeout(5),
             shutdown_timeout: 30,
-            exit: true,
+            exit: false,
             no_http2: false,
             no_signals: false,
             maxconn: 102_400,
@@ -132,7 +131,7 @@ where
 
     /// Set server keep-alive setting.
     ///
-    /// By default keep alive is set to a `Os`.
+    /// By default keep alive is set to a 5 seconds.
     pub fn keep_alive<T: Into<KeepAlive>>(mut self, val: T) -> Self {
         self.keep_alive = val.into();
         self
@@ -637,7 +636,9 @@ where
 
     fn shutdown(&self, force: bool) {
         if force {
-            self.settings.head().traverse::<TcpStream, H>();
+            self.settings
+                .head()
+                .traverse(|ch: &mut HttpChannel<TcpStream, H>| ch.shutdown());
         }
     }
 }
@@ -693,7 +694,7 @@ where
         };
         let _ = io.set_nodelay(true);
 
-        current_thread::spawn(HttpChannel::new(h, io, peer));
+        Arbiter::spawn(HttpChannel::new(h, io, peer));
     }
 }
 
@@ -753,10 +754,10 @@ where
         let _ = io.set_nodelay(true);
 
         let rate = h.connection_rate();
-        current_thread::spawn(self.acceptor.accept(io).then(move |res| {
+        Arbiter::spawn(self.acceptor.accept(io).then(move |res| {
             drop(rate);
             match res {
-                Ok(io) => current_thread::spawn(HttpChannel::new(h, io, peer)),
+                Ok(io) => Arbiter::spawn(HttpChannel::new(h, io, peer)),
                 Err(err) => trace!("Can not establish connection: {}", err),
             }
             Ok(())

src/server/settings.rs

@@ -13,7 +13,7 @@ use http::StatusCode;
 use lazycell::LazyCell;
 use parking_lot::Mutex;
 use time;
-use tokio_timer::Interval;
+use tokio_timer::{Delay, Interval};
 
 use super::channel::Node;
 use super::message::{Request, RequestPool};
@@ -197,6 +197,16 @@ impl<H> WorkerSettings<H> {
         &self.h
     }
 
+    pub fn keep_alive_timer(&self) -> Option<Delay> {
+        if self.keep_alive != 0 {
+            Some(Delay::new(
+                Instant::now() + Duration::from_secs(self.keep_alive),
+            ))
+        } else {
+            None
+        }
+    }
+
     pub fn keep_alive(&self) -> u64 {
         self.keep_alive
     }

src/test.rs

@@ -120,6 +120,7 @@ impl TestServer {
             HttpServer::new(factory)
                 .disable_signals()
                 .listen(tcp)
+                .keep_alive(5)
                 .start();
 
             tx.send((System::current(), local_addr, TestServer::get_conn()))
@@ -328,6 +329,7 @@ impl<S: 'static> TestServerBuilder<S> {
                 config(&mut app);
                 vec![app]
             }).workers(1)
+            .keep_alive(5)
             .disable_signals();
 
             tx.send((System::current(), addr, TestServer::get_conn()))

tests/test_client.rs

@@ -407,24 +407,29 @@ fn test_client_cookie_handling() {
         let cookie2 = cookie2b.clone();
         app.handler(move |req: &HttpRequest| {
             // Check cookies were sent correctly
-            req.cookie("cookie1").ok_or_else(err)
+            req.cookie("cookie1")
-                    .and_then(|c1| if c1.value() == "value1" {
+                .ok_or_else(err)
+                .and_then(|c1| {
+                    if c1.value() == "value1" {
                         Ok(())
                     } else {
                         Err(err())
-                    })
+                    }
-                    .and_then(|()| req.cookie("cookie2").ok_or_else(err))
+                }).and_then(|()| req.cookie("cookie2").ok_or_else(err))
-                    .and_then(|c2| if c2.value() == "value2" {
+                .and_then(|c2| {
+                    if c2.value() == "value2" {
                         Ok(())
                     } else {
                         Err(err())
-                    })
+                    }
-                    // Send some cookies back
+                })
-                    .map(|_| HttpResponse::Ok()
+                // Send some cookies back
-                         .cookie(cookie1.clone())
+                .map(|_| {
-                         .cookie(cookie2.clone())
+                    HttpResponse::Ok()
-                         .finish()
+                        .cookie(cookie1.clone())
-                    )
+                        .cookie(cookie2.clone())
+                        .finish()
+                })
         })
     });