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

Fix issue with HttpChannel linked list.

CHANGES.md

@@ -1,5 +1,29 @@
 # 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

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-web"
-version = "0.7.6"
+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"

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());
-        let _ = writeln!(f, "  headers:");
+        writeln!(f, "\nClientResponse {:?} {}", self.version(), self.status())?;
+        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/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);
-        let _ = writeln!(f, "  boundary: {}", self.inner.borrow().boundary);
-        let _ = writeln!(f, "  headers:");
+        writeln!(f, "\nMultipartField: {}", self.ct)?;
+        writeln!(f, "  boundary: {}", self.inner.borrow().boundary)?;
+        writeln!(f, "  headers:")?;
         for (key, val) in self.headers.iter() {
-            let _ = writeln!(f, "    {:?}: {:?}", key, val);
+            writeln!(f, "    {:?}: {:?}", key, val)?;
         }
-        res
+        Ok(())
     }
 }
 

src/router.rs

@@ -832,7 +832,7 @@ impl ResourceDef {
             }).expect("malformed param");
         let (mut param, rem) = pattern.split_at(close_idx + 1);
         param = &param[1..param.len() - 1]; // Remove outer brackets
-        let (name, pattern) = match param.find(":") {
+        let (name, pattern) = match param.find(':') {
             Some(idx) => {
                 let (name, pattern) = param.split_at(idx);
                 (name, &pattern[1..])
@@ -849,7 +849,7 @@ impl ResourceDef {
     fn parse(
         mut pattern: &str, for_prefix: bool,
     ) -> (String, Vec<PatternElement>, bool, usize) {
-        if pattern.find("{").is_none() {
+        if pattern.find('{').is_none() {
             return (
                 String::from(pattern),
                 vec![PatternElement::Str(String::from(pattern))],
@@ -861,7 +861,7 @@ impl ResourceDef {
         let mut elems = Vec::new();
         let mut re = String::from("^");
 
-        while let Some(idx) = pattern.find("{") {
+        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));

src/scope.rs

@@ -821,11 +821,9 @@ mod tests {
                 scope
                     .route("/path1", Method::GET, |_: HttpRequest<_>| {
                         HttpResponse::Ok()
-                    }).route(
-                        "/path1",
-                        Method::DELETE,
-                        |_: HttpRequest<_>| HttpResponse::Ok(),
-                    )
+                    }).route("/path1", 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 |_| {
-                                        let _ = r.set_readiness(mio::Ready::readable());
+                                    .and_then(
+                                        move |_| {
+                                            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();
-                match result {
-                    Ok(Async::Ready(())) | Err(_) => {
-                        if let Some(n) = self.node.as_mut() {
-                            n.remove()
-                        };
-                    }
-                    _ => (),
-                }
-                return result;
+                return h1.poll();
             }
             Some(HttpProtocol::H2(ref mut h2)) => {
-                let result = h2.poll();
-                match result {
-                    Ok(Async::Ready(())) | Err(_) => {
-                        if let Some(n) = self.node.as_mut() {
-                            n.remove()
-                        };
-                    }
-                    _ => (),
-                }
-                return result;
+                return h2.poll();
             }
             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

@@ -3,12 +3,11 @@ use std::rc::Rc;
 use std::sync::Arc;
 use std::{io, mem, net, time};
 
-use actix::{Actor, Addr, AsyncContext, Context, Handler, System};
+use actix::{Actor, Addr, Arbiter, AsyncContext, Context, Handler, System};
 
 use futures::{Future, Stream};
 use net2::{TcpBuilder, TcpStreamExt};
 use num_cpus;
-use tokio_current_thread::spawn;
 use tokio_io::{AsyncRead, AsyncWrite};
 use tokio_tcp::TcpStream;
 
@@ -71,7 +70,7 @@ where
             factory: Arc::new(f),
             host: None,
             backlog: 2048,
-            keep_alive: KeepAlive::Os,
+            keep_alive: KeepAlive::Timeout(5),
             shutdown_timeout: 30,
             exit: false,
             no_http2: false,
@@ -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
@@ -585,7 +584,7 @@ where
     type Result = ();
 
     fn handle(&mut self, msg: Conn<T>, _: &mut Context<Self>) -> Self::Result {
-        spawn(HttpChannel::new(
+        Arbiter::spawn(HttpChannel::new(
             Rc::clone(&self.settings),
             msg.io,
             msg.peer,
@@ -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);
 
-        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();
-        spawn(self.acceptor.accept(io).then(move |res| {
+        Arbiter::spawn(self.acceptor.accept(io).then(move |res| {
             drop(rate);
             match res {
-                Ok(io) => 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)
-                    .and_then(|c1| if c1.value() == "value1" {
+            req.cookie("cookie1")
+                .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(|c2| if c2.value() == "value2" {
+                    }
+                }).and_then(|()| req.cookie("cookie2").ok_or_else(err))
+                .and_then(|c2| {
+                    if c2.value() == "value2" {
                         Ok(())
                     } else {
                         Err(err())
+                    }
                 })
                 // Send some cookies back
-                    .map(|_| HttpResponse::Ok()
+                .map(|_| {
+                    HttpResponse::Ok()
                         .cookie(cookie1.clone())
                         .cookie(cookie2.clone())
                         .finish()
-                    )
+                })
         })
     });