backport #215

Let CSRF's `allowed_origin()` be specified as a type supporting `Into<String>`

CHANGES.md

@@ -1,5 +1,21 @@
 # Changes
 
+## 0.5.7 (2018-05-09)
+
+* Fix http/2 payload streaming #215
+
+
+## 0.5.6 (2018-04-24)
+
+* Make flate2 crate optional #200
+
+
+## 0.5.5 (2018-04-24)
+
+* Fix panic when Websocket is closed with no error code #191
+
+* Allow to use rust backend for flate2 crate #199
+
 ## 0.5.4 (2018-04-19)
 
 * Add identity service middleware

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-web"
-version = "0.5.4"
+version = "0.5.7"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix web is a simple, pragmatic and extremely fast web framework for Rust."
 readme = "README.md"
@@ -26,7 +26,7 @@ name = "actix_web"
 path = "src/lib.rs"
 
 [features]
-default = ["session", "brotli"]
+default = ["session", "brotli", "flate2-c"]
 
 # tls
 tls = ["native-tls", "tokio-tls"]
@@ -34,19 +34,24 @@ tls = ["native-tls", "tokio-tls"]
 # openssl
 alpn = ["openssl", "tokio-openssl"]
 
-# sessions
+# sessions feature, session require "ring" crate and c compiler
 session = ["cookie/secure"]
 
-# brotli encoding
+# brotli encoding, requires c compiler
 brotli = ["brotli2"]
 
+# miniz-sys backend for flate2 crate
+flate2-c = ["flate2/miniz-sys"]
+
+# rust backend for flate2 crate
+flate2-rust = ["flate2/rust_backend"]
+
 [dependencies]
 actix = "^0.5.5"
 
 base64 = "0.9"
 bitflags = "1.0"
 failure = "0.1.1"
-flate2 = "1.0"
 h2 = "0.1"
 http = "^0.1.5"
 httparse = "1.2"
@@ -71,6 +76,7 @@ lazy_static = "1.0"
 url = { version="1.7", features=["query_encoding"] }
 cookie = { version="0.10", features=["percent-encode"] }
 brotli2 = { version="^0.3.2", optional = true }
+flate2 = { version="1.0", optional = true, default-features = false }
 
 # io
 mio = "^0.6.13"

README.md

@@ -2,12 +2,12 @@
 
 Actix web is a simple, pragmatic and extremely fast web framework for Rust.
 
-* Supported *HTTP/1.x* and [*HTTP/2.0*](https://actix.rs/actix-web/guide/qs_13.html) protocols
+* Supported *HTTP/1.x* and [*HTTP/2.0*](https://actix.rs/book/actix-web/sec-12-http2.html) protocols
 * Streaming and pipelining
 * Keep-alive and slow requests handling
-* Client/server [WebSockets](https://actix.rs/actix-web/guide/qs_9.html) support
+* Client/server [WebSockets](https://actix.rs/book/actix-web/sec-11-websockets.html) support
 * Transparent content compression/decompression (br, gzip, deflate)
-* Configurable [request routing](https://actix.rs/actix-web/guide/qs_5.html)
+* Configurable [request routing](https://actix.rs/book/actix-web/sec-6-url-dispatch.html)
 * Graceful server shutdown
 * Multipart streams
 * Static assets
@@ -54,9 +54,11 @@ fn main() {
 * [Stateful](https://github.com/actix/examples/tree/master/state/)
 * [Protobuf support](https://github.com/actix/examples/tree/master/protobuf/)
 * [Multipart streams](https://github.com/actix/examples/tree/master/multipart/)
-* [Simple websocket session](https://github.com/actix/examples/tree/master/websocket/)
-* [Tera templates](https://github.com/actix/examples/tree/master/template_tera/)
+* [Simple websocket](https://github.com/actix/examples/tree/master/websocket/)
+* [Tera](https://github.com/actix/examples/tree/master/template_tera/) / 
+  [Askama](https://github.com/actix/examples/tree/master/template_askama/) templates
 * [Diesel integration](https://github.com/actix/examples/tree/master/diesel/)
+* [r2d2](https://github.com/actix/examples/tree/master/r2d2/)
 * [SSL / HTTP/2.0](https://github.com/actix/examples/tree/master/tls/)
 * [Tcp/Websocket chat](https://github.com/actix/examples/tree/master/websocket-chat/)
 * [Json](https://github.com/actix/examples/tree/master/json/)

src/client/writer.rs

@@ -7,8 +7,10 @@ use std::io::{self, Write};
 #[cfg(feature = "brotli")]
 use brotli2::write::BrotliEncoder;
 use bytes::{BufMut, BytesMut};
-use flate2::Compression;
+#[cfg(feature = "flate2")]
 use flate2::write::{DeflateEncoder, GzEncoder};
+#[cfg(feature = "flate2")]
+use flate2::Compression;
 use futures::{Async, Poll};
 use http::header::{HeaderValue, CONNECTION, CONTENT_ENCODING, CONTENT_LENGTH, DATE,
                    TRANSFER_ENCODING};
@@ -18,9 +20,9 @@ use tokio_io::AsyncWrite;
 
 use body::{Binary, Body};
 use header::ContentEncoding;
-use server::WriterState;
 use server::encoding::{ContentEncoder, TransferEncoding};
 use server::shared::SharedBytes;
+use server::WriterState;
 
 use client::ClientRequest;
 
@@ -70,7 +72,7 @@ impl HttpClientWriter {
     // !self.flags.contains(Flags::UPGRADE) }
 
     fn write_to_stream<T: AsyncWrite>(
-        &mut self, stream: &mut T
+        &mut self, stream: &mut T,
     ) -> io::Result<WriterState> {
         while !self.buffer.is_empty() {
             match stream.write(self.buffer.as_ref()) {
@@ -191,7 +193,7 @@ impl HttpClientWriter {
 
     #[inline]
     pub fn poll_completed<T: AsyncWrite>(
-        &mut self, stream: &mut T, shutdown: bool
+        &mut self, stream: &mut T, shutdown: bool,
     ) -> Poll<(), io::Error> {
         match self.write_to_stream(stream) {
             Ok(WriterState::Done) => {
@@ -222,9 +224,11 @@ fn content_encoder(buf: SharedBytes, req: &mut ClientRequest) -> ContentEncoder
                 let tmp = SharedBytes::default();
                 let transfer = TransferEncoding::eof(tmp.clone());
                 let mut enc = match encoding {
+                    #[cfg(feature = "flate2")]
                     ContentEncoding::Deflate => ContentEncoder::Deflate(
                         DeflateEncoder::new(transfer, Compression::default()),
                     ),
+                    #[cfg(feature = "flate2")]
                     ContentEncoding::Gzip => ContentEncoder::Gzip(GzEncoder::new(
                         transfer,
                         Compression::default(),
@@ -283,10 +287,12 @@ fn content_encoder(buf: SharedBytes, req: &mut ClientRequest) -> ContentEncoder
 
     req.replace_body(body);
     match encoding {
+        #[cfg(feature = "flate2")]
         ContentEncoding::Deflate => ContentEncoder::Deflate(DeflateEncoder::new(
             transfer,
             Compression::default(),
         )),
+        #[cfg(feature = "flate2")]
         ContentEncoding::Gzip => {
             ContentEncoder::Gzip(GzEncoder::new(transfer, Compression::default()))
         }
@@ -299,7 +305,7 @@ fn content_encoder(buf: SharedBytes, req: &mut ClientRequest) -> ContentEncoder
 }
 
 fn streaming_encoding(
-    buf: SharedBytes, version: Version, req: &mut ClientRequest
+    buf: SharedBytes, version: Version, req: &mut ClientRequest,
 ) -> TransferEncoding {
     if req.chunked() {
         // Enable transfer encoding

src/header/mod.rs

@@ -6,8 +6,8 @@ use std::str::FromStr;
 
 use bytes::{Bytes, BytesMut};
 use mime::Mime;
-use modhttp::Error as HttpError;
 use modhttp::header::GetAll;
+use modhttp::Error as HttpError;
 
 pub use modhttp::header::*;
 
@@ -116,8 +116,10 @@ pub enum ContentEncoding {
     #[cfg(feature = "brotli")]
     Br,
     /// A format using the zlib structure with deflate algorithm
+    #[cfg(feature = "flate2")]
     Deflate,
     /// Gzip algorithm
+    #[cfg(feature = "flate2")]
     Gzip,
     /// Indicates the identity function (i.e. no compression, nor modification)
     Identity,
@@ -137,7 +139,9 @@ impl ContentEncoding {
         match *self {
             #[cfg(feature = "brotli")]
             ContentEncoding::Br => "br",
+            #[cfg(feature = "flate2")]
             ContentEncoding::Gzip => "gzip",
+            #[cfg(feature = "flate2")]
             ContentEncoding::Deflate => "deflate",
             ContentEncoding::Identity | ContentEncoding::Auto => "identity",
         }
@@ -149,7 +153,9 @@ impl ContentEncoding {
         match *self {
             #[cfg(feature = "brotli")]
             ContentEncoding::Br => 1.1,
+            #[cfg(feature = "flate2")]
             ContentEncoding::Gzip => 1.0,
+            #[cfg(feature = "flate2")]
             ContentEncoding::Deflate => 0.9,
             ContentEncoding::Identity | ContentEncoding::Auto => 0.1,
         }
@@ -159,10 +165,12 @@ impl ContentEncoding {
 // TODO: remove memory allocation
 impl<'a> From<&'a str> for ContentEncoding {
     fn from(s: &'a str) -> ContentEncoding {
-        match s.trim().to_lowercase().as_ref() {
+        match AsRef::<str>::as_ref(&s.trim().to_lowercase()) {
             #[cfg(feature = "brotli")]
             "br" => ContentEncoding::Br,
+            #[cfg(feature = "flate2")]
             "gzip" => ContentEncoding::Gzip,
+            #[cfg(feature = "flate2")]
             "deflate" => ContentEncoding::Deflate,
             _ => ContentEncoding::Identity,
         }
@@ -202,7 +210,7 @@ impl fmt::Write for Writer {
 #[doc(hidden)]
 /// Reads a comma-delimited raw header into a Vec.
 pub fn from_comma_delimited<T: FromStr>(
-    all: GetAll<HeaderValue>
+    all: GetAll<HeaderValue>,
 ) -> Result<Vec<T>, ParseError> {
     let mut result = Vec::new();
     for h in all {

src/httprequest.rs

@@ -37,6 +37,7 @@ pub struct HttpInnerMessage {
     pub addr: Option<SocketAddr>,
     pub payload: Option<Payload>,
     pub info: Option<ConnectionInfo<'static>>,
+    pub keep_alive: bool,
     resource: RouterResource,
 }
 
@@ -56,11 +57,12 @@ impl Default for HttpInnerMessage {
             params: Params::new(),
             query: Params::new(),
             query_loaded: false,
-            cookies: None,
             addr: None,
+            cookies: None,
             payload: None,
             extensions: Extensions::new(),
             info: None,
+            keep_alive: true,
             resource: RouterResource::Notset,
         }
     }
@@ -70,20 +72,7 @@ impl HttpInnerMessage {
     /// Checks if a connection should be kept alive.
     #[inline]
     pub fn keep_alive(&self) -> bool {
-        if let Some(conn) = self.headers.get(header::CONNECTION) {
-            if let Ok(conn) = conn.to_str() {
-                if self.version == Version::HTTP_10 && conn.contains("keep-alive") {
-                    true
-                } else {
-                    self.version == Version::HTTP_11
-                        && !(conn.contains("close") || conn.contains("upgrade"))
-                }
-            } else {
-                false
-            }
-        } else {
-            self.version != Version::HTTP_10
-        }
+        self.keep_alive
     }
 
     #[inline]
@@ -91,12 +80,12 @@ impl HttpInnerMessage {
         self.headers.clear();
         self.extensions.clear();
         self.params.clear();
-        self.query.clear();
-        self.query_loaded = false;
-        self.cookies = None;
         self.addr = None;
         self.info = None;
+        self.query_loaded = false;
+        self.cookies = None;
         self.payload = None;
+        self.keep_alive = true;
         self.resource = RouterResource::Notset;
     }
 }
@@ -126,10 +115,11 @@ impl HttpRequest<()> {
                 params: Params::new(),
                 query: Params::new(),
                 query_loaded: false,
+                extensions: Extensions::new(),
                 cookies: None,
                 addr: None,
-                extensions: Extensions::new(),
                 info: None,
+                keep_alive: true,
                 resource: RouterResource::Notset,
             }),
             None,
@@ -377,13 +367,13 @@ impl<S> HttpRequest<S> {
     /// To get client connection information `connection_info()` method should
     /// be used.
     #[inline]
-    pub fn peer_addr(&self) -> Option<&SocketAddr> {
-        self.as_ref().addr.as_ref()
+    pub fn peer_addr(&self) -> Option<SocketAddr> {
+        self.as_ref().addr
     }
 
     #[inline]
     pub(crate) fn set_peer_addr(&mut self, addr: Option<SocketAddr>) {
-        self.as_mut().addr = addr
+        self.as_mut().addr = addr;
     }
 
     /// Get a reference to the Params object.
@@ -392,6 +382,7 @@ impl<S> HttpRequest<S> {
         if !self.as_ref().query_loaded {
             let params: &mut Params =
                 unsafe { mem::transmute(&mut self.as_mut().query) };
+            params.clear();
             self.as_mut().query_loaded = true;
             for (key, val) in form_urlencoded::parse(self.query_string().as_ref()) {
                 params.add(key, val);
@@ -425,9 +416,9 @@ impl<S> HttpRequest<S> {
                     }
                 }
             }
-            msg.cookies = Some(cookies)
+            msg.cookies = Some(cookies);
         }
-        Ok(self.as_ref().cookies.as_ref().unwrap())
+        Ok(&self.as_ref().cookies.as_ref().unwrap())
     }
 
     /// Return request cookie.

src/lib.rs

@@ -64,8 +64,10 @@
 #![cfg_attr(actix_nightly, feature(
     specialization, // for impl ErrorResponse for std::error::Error
 ))]
-#![cfg_attr(feature = "cargo-clippy",
-            allow(decimal_literal_representation, suspicious_arithmetic_impl))]
+#![cfg_attr(
+    feature = "cargo-clippy",
+    allow(decimal_literal_representation, suspicious_arithmetic_impl)
+)]
 
 #[macro_use]
 extern crate log;
@@ -103,6 +105,7 @@ extern crate serde;
 #[cfg(feature = "brotli")]
 extern crate brotli2;
 extern crate encoding;
+#[cfg(feature = "flate2")]
 extern crate flate2;
 extern crate h2 as http2;
 extern crate num_cpus;

src/middleware/csrf.rs

@@ -150,8 +150,8 @@ impl CsrfFilter {
 
     /// Add an origin that is allowed to make requests. Will be verified
     /// against the `Origin` request header.
-    pub fn allowed_origin(mut self, origin: &str) -> CsrfFilter {
-        self.origins.insert(origin.to_owned());
+    pub fn allowed_origin<T: Into<String>>(mut self, origin: T) -> CsrfFilter {
+        self.origins.insert(origin.into());
         self
     }
 

src/pipeline.rs

@@ -491,8 +491,8 @@ impl<S: 'static, H> ProcessResponse<S, H> {
                             if let Some(err) = self.resp.error() {
                                 if self.resp.status().is_server_error() {
                                     error!(
-                                        "Error occured during request handling: {}",
-                                        err
+                                        "Error occured during request handling, status: {} {}",
+                                        self.resp.status(), err
                                     );
                                 } else {
                                     warn!(

src/server/encoding.rs

@@ -6,11 +6,14 @@ use std::{cmp, io, mem};
 #[cfg(feature = "brotli")]
 use brotli2::write::{BrotliDecoder, BrotliEncoder};
 use bytes::{BufMut, Bytes, BytesMut};
-use flate2::Compression;
+#[cfg(feature = "flate2")]
 use flate2::read::GzDecoder;
+#[cfg(feature = "flate2")]
 use flate2::write::{DeflateDecoder, DeflateEncoder, GzEncoder};
-use http::header::{HeaderMap, HeaderValue, ACCEPT_ENCODING, CONNECTION,
-                   CONTENT_ENCODING, CONTENT_LENGTH, TRANSFER_ENCODING};
+#[cfg(feature = "flate2")]
+use flate2::Compression;
+use http::header::{HeaderMap, HeaderValue, ACCEPT_ENCODING, CONTENT_ENCODING,
+                   CONTENT_LENGTH, TRANSFER_ENCODING};
 use http::{HttpTryFrom, Method, Version};
 
 use body::{Binary, Body};
@@ -144,7 +147,9 @@ impl PayloadWriter for EncodedPayload {
 }
 
 pub(crate) enum Decoder {
+    #[cfg(feature = "flate2")]
     Deflate(Box<DeflateDecoder<Writer>>),
+    #[cfg(feature = "flate2")]
     Gzip(Option<Box<GzDecoder<Wrapper>>>),
     #[cfg(feature = "brotli")]
     Br(Box<BrotliDecoder<Writer>>),
@@ -223,9 +228,11 @@ impl PayloadStream {
             ContentEncoding::Br => {
                 Decoder::Br(Box::new(BrotliDecoder::new(Writer::new())))
             }
+            #[cfg(feature = "flate2")]
             ContentEncoding::Deflate => {
                 Decoder::Deflate(Box::new(DeflateDecoder::new(Writer::new())))
             }
+            #[cfg(feature = "flate2")]
             ContentEncoding::Gzip => Decoder::Gzip(None),
             _ => Decoder::Identity,
         };
@@ -251,6 +258,7 @@ impl PayloadStream {
                 }
                 Err(e) => Err(e),
             },
+            #[cfg(feature = "flate2")]
             Decoder::Gzip(ref mut decoder) => {
                 if let Some(ref mut decoder) = *decoder {
                     decoder.as_mut().get_mut().eof = true;
@@ -267,6 +275,7 @@ impl PayloadStream {
                     Ok(None)
                 }
             }
+            #[cfg(feature = "flate2")]
             Decoder::Deflate(ref mut decoder) => match decoder.try_finish() {
                 Ok(_) => {
                     let b = decoder.get_mut().take();
@@ -297,6 +306,7 @@ impl PayloadStream {
                 }
                 Err(e) => Err(e),
             },
+            #[cfg(feature = "flate2")]
             Decoder::Gzip(ref mut decoder) => {
                 if decoder.is_none() {
                     *decoder = Some(Box::new(GzDecoder::new(Wrapper {
@@ -334,6 +344,7 @@ impl PayloadStream {
                     }
                 }
             }
+            #[cfg(feature = "flate2")]
             Decoder::Deflate(ref mut decoder) => match decoder.write_all(&data) {
                 Ok(_) => {
                     decoder.flush()?;
@@ -352,7 +363,9 @@ impl PayloadStream {
 }
 
 pub(crate) enum ContentEncoder {
+    #[cfg(feature = "flate2")]
     Deflate(DeflateEncoder<TransferEncoding>),
+    #[cfg(feature = "flate2")]
     Gzip(GzEncoder<TransferEncoding>),
     #[cfg(feature = "brotli")]
     Br(BrotliEncoder<TransferEncoding>),
@@ -422,9 +435,11 @@ impl ContentEncoder {
                     let tmp = SharedBytes::default();
                     let transfer = TransferEncoding::eof(tmp.clone());
                     let mut enc = match encoding {
+                        #[cfg(feature = "flate2")]
                         ContentEncoding::Deflate => ContentEncoder::Deflate(
                             DeflateEncoder::new(transfer, Compression::fast()),
                         ),
+                        #[cfg(feature = "flate2")]
                         ContentEncoding::Gzip => ContentEncoder::Gzip(GzEncoder::new(
                             transfer,
                             Compression::fast(),
@@ -459,9 +474,6 @@ impl ContentEncoder {
                 if resp.upgrade() {
                     if version == Version::HTTP_2 {
                         error!("Connection upgrade is forbidden for HTTP/2");
-                    } else {
-                        resp.headers_mut()
-                            .insert(CONNECTION, HeaderValue::from_static("upgrade"));
                     }
                     if encoding != ContentEncoding::Identity {
                         encoding = ContentEncoding::Identity;
@@ -481,10 +493,12 @@ impl ContentEncoder {
         }
 
         match encoding {
+            #[cfg(feature = "flate2")]
             ContentEncoding::Deflate => ContentEncoder::Deflate(DeflateEncoder::new(
                 transfer,
                 Compression::fast(),
             )),
+            #[cfg(feature = "flate2")]
             ContentEncoding::Gzip => {
                 ContentEncoder::Gzip(GzEncoder::new(transfer, Compression::fast()))
             }
@@ -497,7 +511,7 @@ impl ContentEncoder {
     }
 
     fn streaming_encoding(
-        buf: SharedBytes, version: Version, resp: &mut HttpResponse
+        buf: SharedBytes, version: Version, resp: &mut HttpResponse,
     ) -> TransferEncoding {
         match resp.chunked() {
             Some(true) => {
@@ -566,7 +580,9 @@ impl ContentEncoder {
         match *self {
             #[cfg(feature = "brotli")]
             ContentEncoder::Br(ref encoder) => encoder.get_ref().is_eof(),
+            #[cfg(feature = "flate2")]
             ContentEncoder::Deflate(ref encoder) => encoder.get_ref().is_eof(),
+            #[cfg(feature = "flate2")]
             ContentEncoder::Gzip(ref encoder) => encoder.get_ref().is_eof(),
             ContentEncoder::Identity(ref encoder) => encoder.is_eof(),
         }
@@ -590,6 +606,7 @@ impl ContentEncoder {
                 }
                 Err(err) => Err(err),
             },
+            #[cfg(feature = "flate2")]
             ContentEncoder::Gzip(encoder) => match encoder.finish() {
                 Ok(mut writer) => {
                     writer.encode_eof();
@@ -598,6 +615,7 @@ impl ContentEncoder {
                 }
                 Err(err) => Err(err),
             },
+            #[cfg(feature = "flate2")]
             ContentEncoder::Deflate(encoder) => match encoder.finish() {
                 Ok(mut writer) => {
                     writer.encode_eof();
@@ -628,6 +646,7 @@ impl ContentEncoder {
                     }
                 }
             }
+            #[cfg(feature = "flate2")]
             ContentEncoder::Gzip(ref mut encoder) => {
                 match encoder.write_all(data.as_ref()) {
                     Ok(_) => Ok(()),
@@ -637,6 +656,7 @@ impl ContentEncoder {
                     }
                 }
             }
+            #[cfg(feature = "flate2")]
             ContentEncoder::Deflate(ref mut encoder) => {
                 match encoder.write_all(data.as_ref()) {
                     Ok(_) => Ok(()),

src/server/h1.rs

@@ -510,9 +510,10 @@ impl Reader {
         buf: &mut BytesMut, settings: &WorkerSettings<H>
     ) -> Poll<(HttpRequest, Option<PayloadInfo>), ParseError> {
         // Parse http message
-        let mut has_te = false;
         let mut has_upgrade = false;
-        let mut has_length = false;
+        let mut chunked = false;
+        let mut content_length = None;
+
         let msg = {
             let bytes_ptr = buf.as_ref().as_ptr() as usize;
             let mut headers: [httparse::Header; MAX_HEADERS] =
@@ -546,10 +547,10 @@ impl Reader {
             let msg = settings.get_http_message();
             {
                 let msg_mut = msg.get_mut();
+                msg_mut.keep_alive = version != Version::HTTP_10;
+
                 for header in headers[..headers_len].iter() {
                     if let Ok(name) = HeaderName::from_bytes(header.name.as_bytes()) {
-                        has_te = has_te || name == header::TRANSFER_ENCODING;
-                        has_length = has_length || name == header::CONTENT_LENGTH;
                         has_upgrade = has_upgrade || name == header::UPGRADE;
                         let v_start = header.value.as_ptr() as usize - bytes_ptr;
                         let v_end = v_start + header.value.len();
@@ -558,6 +559,47 @@ impl Reader {
                                 slice.slice(v_start, v_end),
                             )
                         };
+                        match name {
+                            header::CONTENT_LENGTH => {
+                                if let Ok(s) = value.to_str() {
+                                    if let Ok(len) = s.parse::<u64>() {
+                                        content_length = Some(len)
+                                    } else {
+                                        debug!("illegal Content-Length: {:?}", len);
+                                        return Err(ParseError::Header);
+                                    }
+                                } else {
+                                    debug!("illegal Content-Length: {:?}", len);
+                                    return Err(ParseError::Header);
+                                }
+                            },
+                            // transfer-encoding
+                            header::TRANSFER_ENCODING => {
+                                if let Ok(s) = value.to_str() {
+                                    chunked = s.to_lowercase().contains("chunked");
+                                } else {
+                                    return Err(ParseError::Header)
+                                }
+                            },
+                            // connection keep-alive state
+                            header::CONNECTION => {
+                                msg_mut.keep_alive = if let Ok(conn) = value.to_str() {
+                                    if version == Version::HTTP_10
+                                        && conn.contains("keep-alive")
+                                    {
+                                        true
+                                    } else {
+                                        version == Version::HTTP_11
+                                            && !(conn.contains("close")
+                                                 || conn.contains("upgrade"))
+                                    }
+                                } else {
+                                    false
+                                };
+                            },
+                            _ => (),
+                        }
+
                         msg_mut.headers.append(name, value);
                     } else {
                         return Err(ParseError::Header);
@@ -572,26 +614,12 @@ impl Reader {
         };
 
         // https://tools.ietf.org/html/rfc7230#section-3.3.3
-        let decoder = if has_te && chunked(&msg.get_mut().headers)? {
+        let decoder = if chunked {
             // Chunked encoding
             Some(Decoder::chunked())
-        } else if has_length {
+        } else if let Some(len) = content_length {
             // Content-Length
-            let len = msg.get_ref()
-                .headers
-                .get(header::CONTENT_LENGTH)
-                .unwrap();
-            if let Ok(s) = len.to_str() {
-                if let Ok(len) = s.parse::<u64>() {
-                    Some(Decoder::length(len))
-                } else {
-                    debug!("illegal Content-Length: {:?}", len);
-                    return Err(ParseError::Header);
-                }
-            } else {
-                debug!("illegal Content-Length: {:?}", len);
-                return Err(ParseError::Header);
-            }
+            Some(Decoder::length(len))
         } else if has_upgrade || msg.get_ref().method == Method::CONNECT {
             // upgrade(websocket) or connect
             Some(Decoder::eof())

src/server/h1writer.rs

@@ -2,8 +2,6 @@
 
 use bytes::BufMut;
 use futures::{Async, Poll};
-use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, DATE};
-use http::{Method, Version};
 use std::rc::Rc;
 use std::{io, mem};
 use tokio_io::AsyncWrite;
@@ -17,6 +15,8 @@ use body::{Binary, Body};
 use header::ContentEncoding;
 use httprequest::HttpInnerMessage;
 use httpresponse::HttpResponse;
+use http::{Method, Version};
+use http::header::{HeaderValue, CONNECTION, CONTENT_LENGTH, DATE};
 
 const AVERAGE_HEADER_SIZE: usize = 30; // totally scientific
 

src/server/h2.rs

@@ -343,24 +343,27 @@ impl<H: 'static> Entry<H> {
     }
 
     fn poll_payload(&mut self) {
-        if !self.flags.contains(EntryFlags::REOF) {
-            if self.payload.need_read() == PayloadStatus::Read {
-                if let Err(err) = self.recv.release_capacity().release_capacity(32_768) {
-                    self.payload.set_error(PayloadError::Http2(err))
-                }
-            } else if let Err(err) = self.recv.release_capacity().release_capacity(0) {
-                self.payload.set_error(PayloadError::Http2(err))
-            }
-
+        while !self.flags.contains(EntryFlags::REOF)
+            && self.payload.need_read() == PayloadStatus::Read
+        {
             match self.recv.poll() {
                 Ok(Async::Ready(Some(chunk))) => {
+                    let l = chunk.len();
                     self.payload.feed_data(chunk);
+                    if let Err(err) = self.recv.release_capacity().release_capacity(l) {
+                        self.payload.set_error(PayloadError::Http2(err));
+                        break;
+                    }
                 }
                 Ok(Async::Ready(None)) => {
                     self.flags.insert(EntryFlags::REOF);
+                    self.payload.feed_eof();
+                }
+                Ok(Async::NotReady) => break,
+                Err(err) => {
+                    self.payload.set_error(PayloadError::Http2(err));
+                    break;
                 }
-                Ok(Async::NotReady) => (),
-                Err(err) => self.payload.set_error(PayloadError::Http2(err)),
             }
         }
     }

src/ws/mod.rs

@@ -310,10 +310,15 @@ where
                     }
                     OpCode::Close => {
                         self.closed = true;
-                        let code = NetworkEndian::read_uint(payload.as_ref(), 2) as u16;
-                        Ok(Async::Ready(Some(Message::Close(CloseCode::from(
-                            code,
-                        )))))
+                        let close_code = if payload.len() >= 2 {
+                            let raw_code =
+                                NetworkEndian::read_uint(payload.as_ref(), 2) as u16;
+                            CloseCode::from(raw_code)
+                        } else {
+                            CloseCode::Status
+                        };
+
+                        Ok(Async::Ready(Some(Message::Close(close_code))))
                     }
                     OpCode::Ping => Ok(Async::Ready(Some(Message::Ping(
                         String::from_utf8_lossy(payload.as_ref()).into(),

tests/test_server.rs

@@ -814,7 +814,7 @@ fn test_h2() {
             })
         });
     let _res = core.run(tcp);
-    // assert_eq!(res.unwrap(), Bytes::from_static(STR.as_ref()));
+    // assert_eq!(_res.unwrap(), Bytes::from_static(STR.as_ref()));
 }
 
 #[test]

tests/test_ws.rs

@@ -60,6 +60,16 @@ fn test_simple() {
     assert_eq!(item, Some(ws::Message::Close(ws::CloseCode::Normal)));
 }
 
+#[test]
+fn test_empty_close_code() {
+    let mut srv = test::TestServer::new(|app| app.handler(|req| ws::start(req, Ws)));
+    let (reader, mut writer) = srv.ws().unwrap();
+
+    writer.close(ws::CloseCode::Empty, "");
+    let (item, _) = srv.execute(reader.into_future()).unwrap();
+    assert_eq!(item, Some(ws::Message::Close(ws::CloseCode::Status)));
+}
+
 #[test]
 fn test_large_text() {
     let data = rand::thread_rng()