prep release

Use str::repeat

CHANGES.md

@@ -1,5 +1,20 @@
 # Changes
 
+## 0.4.4 (2018-03-04)
+
+* Allow to use Arc<Vec<u8>> as response/request body
+
+* Fix handling of requests with an encoded body with a length > 8192 #93
+
+## 0.4.3 (2018-03-03)
+
+* Fix request body read bug
+
+* Fix segmentation fault #79
+
+* Set reuse address before bind #90
+
+
 ## 0.4.2 (2018-03-02)
 
 * Better naming for websockets implementation

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-web"
-version = "0.4.2"
+version = "0.4.4"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix web is a small, pragmatic, extremely fast, web framework for Rust."
 readme = "README.md"

examples/websocket/src/client.rs

@@ -88,7 +88,7 @@ impl Handler<ClientCommand> for ChatClient {
     type Result = ();
 
     fn handle(&mut self, msg: ClientCommand, ctx: &mut Context<Self>) {
-        self.0.text(msg.0.as_str())
+        self.0.text(msg.0)
     }
 }
 

src/body.rs

@@ -36,6 +36,8 @@ pub enum Binary {
     /// Shared string body
     #[doc(hidden)]
     ArcSharedString(Arc<String>),
+    /// Shared vec body
+    SharedVec(Arc<Vec<u8>>),
 }
 
 impl Body {
@@ -115,6 +117,7 @@ impl Binary {
             Binary::Slice(slice) => slice.len(),
             Binary::SharedString(ref s) => s.len(),
             Binary::ArcSharedString(ref s) => s.len(),
+            Binary::SharedVec(ref s) => s.len(),
         }
     }
 
@@ -134,8 +137,9 @@ impl Clone for Binary {
         match *self {
             Binary::Bytes(ref bytes) => Binary::Bytes(bytes.clone()),
             Binary::Slice(slice) => Binary::Bytes(Bytes::from(slice)),
-            Binary::SharedString(ref s) => Binary::Bytes(Bytes::from(s.as_str())),
-            Binary::ArcSharedString(ref s) => Binary::Bytes(Bytes::from(s.as_str())),
+            Binary::SharedString(ref s) => Binary::SharedString(s.clone()),
+            Binary::ArcSharedString(ref s) => Binary::ArcSharedString(s.clone()),
+            Binary::SharedVec(ref s) => Binary::SharedVec(s.clone()),
         }
     }
 }
@@ -147,6 +151,7 @@ impl Into<Bytes> for Binary {
             Binary::Slice(slice) => Bytes::from(slice),
             Binary::SharedString(s) => Bytes::from(s.as_str()),
             Binary::ArcSharedString(s) => Bytes::from(s.as_str()),
+            Binary::SharedVec(s) => Bytes::from(AsRef::<[u8]>::as_ref(s.as_ref())),
         }
     }
 }
@@ -217,6 +222,18 @@ impl<'a> From<&'a Arc<String>> for Binary {
     }
 }
 
+impl From<Arc<Vec<u8>>> for Binary {
+    fn from(body: Arc<Vec<u8>>) -> Binary {
+        Binary::SharedVec(body)
+    }
+}
+
+impl<'a> From<&'a Arc<Vec<u8>>> for Binary {
+    fn from(body: &'a Arc<Vec<u8>>) -> Binary {
+        Binary::SharedVec(Arc::clone(body))
+    }
+}
+
 impl AsRef<[u8]> for Binary {
     fn as_ref(&self) -> &[u8] {
         match *self {
@@ -224,6 +241,7 @@ impl AsRef<[u8]> for Binary {
             Binary::Slice(slice) => slice,
             Binary::SharedString(ref s) => s.as_bytes(),
             Binary::ArcSharedString(ref s) => s.as_bytes(),
+            Binary::SharedVec(ref s) => s.as_ref().as_ref(),
         }
     }
 }
@@ -304,6 +322,15 @@ mod tests {
         assert_eq!(Binary::from(&b).as_ref(), "test".as_bytes());
     }
 
+    #[test]
+    fn test_shared_vec() {
+        let b = Arc::new(Vec::from(&b"test"[..]));
+        assert_eq!(Binary::from(b.clone()).len(), 4);
+        assert_eq!(Binary::from(b.clone()).as_ref(), &b"test"[..]);
+        assert_eq!(Binary::from(&b).len(), 4);
+        assert_eq!(Binary::from(&b).as_ref(), &b"test"[..]);
+    }
+
     #[test]
     fn test_bytes_mut() {
         let b =  BytesMut::from("test");

src/client/parser.rs

@@ -2,7 +2,7 @@ use std::mem;
 use httparse;
 use http::{Version, HttpTryFrom, HeaderMap, StatusCode};
 use http::header::{self, HeaderName, HeaderValue};
-use bytes::{Bytes, BytesMut, BufMut};
+use bytes::{Bytes, BytesMut};
 use futures::{Poll, Async};
 
 use error::{ParseError, PayloadError};
@@ -37,7 +37,7 @@ impl HttpResponseParser {
         where T: IoStream
     {
         // if buf is empty parse_message will always return NotReady, let's avoid that
-        let read = if buf.is_empty() {
+        if buf.is_empty() {
             match utils::read_from_io(io, buf) {
                 Ok(Async::Ready(0)) =>
                     return Err(HttpResponseParserError::Disconnect),
@@ -47,13 +47,12 @@ impl HttpResponseParser {
                 Err(err) =>
                     return Err(HttpResponseParserError::Error(err.into()))
             }
-            false
-        } else {
-            true
-        };
+        }
 
         loop {
-            match HttpResponseParser::parse_message(buf).map_err(HttpResponseParserError::Error)? {
+            match HttpResponseParser::parse_message(buf)
+                .map_err(HttpResponseParserError::Error)?
+            {
                 Async::Ready((msg, decoder)) => {
                     self.decoder = decoder;
                     return Ok(Async::Ready(msg));
@@ -62,17 +61,13 @@ impl HttpResponseParser {
                     if buf.capacity() >= MAX_BUFFER_SIZE {
                         return Err(HttpResponseParserError::Error(ParseError::TooLarge));
                     }
-                    if read || buf.remaining_mut() == 0 {
-                        match utils::read_from_io(io, buf) {
-                            Ok(Async::Ready(0)) =>
-                                return Err(HttpResponseParserError::Disconnect),
-                            Ok(Async::Ready(_)) => (),
-                            Ok(Async::NotReady) => return Ok(Async::NotReady),
-                            Err(err) =>
-                                return Err(HttpResponseParserError::Error(err.into())),
-                        }
-                    } else {
-                        return Ok(Async::NotReady)
+                    match utils::read_from_io(io, buf) {
+                        Ok(Async::Ready(0)) =>
+                            return Err(HttpResponseParserError::Disconnect),
+                        Ok(Async::Ready(_)) => (),
+                        Ok(Async::NotReady) => return Ok(Async::NotReady),
+                        Err(err) =>
+                            return Err(HttpResponseParserError::Error(err.into())),
                     }
                 },
             }
@@ -84,25 +79,34 @@ impl HttpResponseParser {
         where T: IoStream
     {
         if self.decoder.is_some() {
-            // read payload
-            match utils::read_from_io(io, buf) {
-                Ok(Async::Ready(0)) => {
-                    if buf.is_empty() {
-                        return Err(PayloadError::Incomplete)
+            loop {
+                // read payload
+                let not_ready = match utils::read_from_io(io, buf) {
+                    Ok(Async::Ready(0)) => {
+                        if buf.is_empty() {
+                            return Err(PayloadError::Incomplete)
+                        }
+                        true
                     }
-                }
-                Err(err) => return Err(err.into()),
-                _ => (),
-            }
+                    Err(err) => return Err(err.into()),
+                    Ok(Async::NotReady) => true,
+                    _ => false,
+                };
 
-            match self.decoder.as_mut().unwrap().decode(buf) {
-                Ok(Async::Ready(Some(b))) => Ok(Async::Ready(Some(b))),
-                Ok(Async::Ready(None)) => {
-                    self.decoder.take();
-                    Ok(Async::Ready(None))
+                match self.decoder.as_mut().unwrap().decode(buf) {
+                    Ok(Async::Ready(Some(b))) =>
+                        return Ok(Async::Ready(Some(b))),
+                    Ok(Async::Ready(None)) => {
+                        self.decoder.take();
+                        return Ok(Async::Ready(None))
+                    }
+                    Ok(Async::NotReady) => {
+                        if not_ready {
+                            return Ok(Async::NotReady)
+                        }
+                    }
+                    Err(err) => return Err(err.into()),
                 }
-                Ok(Async::NotReady) => Ok(Async::NotReady),
-                Err(err) => Err(err.into()),
             }
         } else {
             Ok(Async::Ready(None))

src/server/channel.rs

@@ -18,15 +18,6 @@ enum HttpProtocol<T: IoStream, H: 'static> {
     Unknown(Rc<WorkerSettings<H>>, Option<SocketAddr>, T, BytesMut),
 }
 
-impl<T: IoStream, H: 'static> HttpProtocol<T, H> {
-    fn is_unknown(&self) -> bool {
-        match *self {
-            HttpProtocol::Unknown(_, _, _, _) => true,
-            _ => false
-        }
-    }
-}
-
 enum ProtocolKind {
     Http1,
     Http2,
@@ -41,18 +32,18 @@ pub struct HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'static {
 impl<T, H> HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'static
 {
     pub(crate) fn new(settings: Rc<WorkerSettings<H>>,
-                      io: T, peer: Option<SocketAddr>, http2: bool) -> HttpChannel<T, H>
+                      mut io: T, peer: Option<SocketAddr>, http2: bool) -> HttpChannel<T, H>
     {
         settings.add_channel();
+        let _ = io.set_nodelay(true);
+
         if http2 {
             HttpChannel {
-                node: None,
-                proto: Some(HttpProtocol::H2(
+                node: None, proto: Some(HttpProtocol::H2(
                     h2::Http2::new(settings, io, peer, Bytes::new()))) }
         } else {
             HttpChannel {
-                node: None,
-                proto: Some(HttpProtocol::Unknown(
+                node: None, proto: Some(HttpProtocol::Unknown(
                     settings, peer, io, BytesMut::with_capacity(4096))) }
         }
     }
@@ -78,15 +69,18 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
     type Error = ();
 
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        if !self.proto.as_ref().map(|p| p.is_unknown()).unwrap_or(false) && self.node.is_none() {
-            self.node = Some(Node::new(self));
-            match self.proto {
+        if !self.node.is_none() {
+            let el = self as *mut _;
+            self.node = Some(Node::new(el));
+            let _ = match self.proto {
                 Some(HttpProtocol::H1(ref mut h1)) =>
-                    h1.settings().head().insert(self.node.as_ref().unwrap()),
+                    self.node.as_ref().map(|n| h1.settings().head().insert(n)),
                 Some(HttpProtocol::H2(ref mut h2)) =>
-                    h2.settings().head().insert(self.node.as_ref().unwrap()),
-                _ => (),
-            }
+                    self.node.as_ref().map(|n| h2.settings().head().insert(n)),
+                Some(HttpProtocol::Unknown(ref mut settings, _, _, _)) =>
+                    self.node.as_ref().map(|n| settings.head().insert(n)),
+                None => unreachable!(),
+            };
         }
 
         let kind = match self.proto {
@@ -95,7 +89,7 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
                 match result {
                     Ok(Async::Ready(())) | Err(_) => {
                         h1.settings().remove_channel();
-                        self.node.as_ref().unwrap().remove();
+                        self.node.as_mut().map(|n| n.remove());
                     },
                     _ => (),
                 }
@@ -106,7 +100,7 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
                 match result {
                     Ok(Async::Ready(())) | Err(_) => {
                         h2.settings().remove_channel();
-                        self.node.as_ref().unwrap().remove();
+                        self.node.as_mut().map(|n| n.remove());
                     },
                     _ => (),
                 }
@@ -117,6 +111,7 @@ impl<T, H> Future for HttpChannel<T, H> where T: IoStream, H: HttpHandler + 'sta
                     Ok(Async::Ready(0)) | Err(_) => {
                         debug!("Ignored premature client disconnection");
                         settings.remove_channel();
+                        self.node.as_mut().map(|n| n.remove());
                         return Err(())
                     },
                     _ => (),
@@ -163,11 +158,11 @@ pub(crate) struct Node<T>
 
 impl<T> Node<T>
 {
-    fn new(el: &mut T) -> Self {
+    fn new(el: *mut T) -> Self {
         Node {
             next: None,
             prev: None,
-            element: el as *mut _,
+            element: el,
         }
     }
 
@@ -186,13 +181,14 @@ impl<T> Node<T>
         }
     }
 
-    fn remove(&self) {
-        #[allow(mutable_transmutes)]
+    fn remove(&mut self) {
         unsafe {
-            if let Some(ref prev) = self.prev {
-                let p: &mut Node<()> = mem::transmute(prev.as_ref().unwrap());
-                let slf: &mut Node<T> = mem::transmute(self);
-                p.next = slf.next.take();
+            self.element = ptr::null_mut();
+            let next = self.next.take();
+            let mut prev = self.prev.take();
+
+            if let Some(ref mut prev) = prev {
+                prev.as_mut().unwrap().next = next;
             }
         }
     }

src/server/encoding.rs

@@ -11,7 +11,7 @@ use flate2::Compression;
 use flate2::read::GzDecoder;
 use flate2::write::{GzEncoder, DeflateDecoder, DeflateEncoder};
 use brotli2::write::{BrotliDecoder, BrotliEncoder};
-use bytes::{Bytes, BytesMut, BufMut, Writer};
+use bytes::{Bytes, BytesMut, BufMut};
 
 use headers::ContentEncoding;
 use body::{Body, Binary};
@@ -128,177 +128,6 @@ impl PayloadWriter for PayloadType {
     }
 }
 
-pub(crate) enum Decoder {
-    Deflate(Box<DeflateDecoder<Writer<BytesMut>>>),
-    Gzip(Option<Box<GzDecoder<Wrapper>>>),
-    Br(Box<BrotliDecoder<Writer<BytesMut>>>),
-    Identity,
-}
-
-// should go after write::GzDecoder get implemented
-#[derive(Debug)]
-pub(crate) struct Wrapper {
-    pub buf: BytesMut,
-    pub eof: bool,
-}
-
-impl io::Read for Wrapper {
-    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        let len = cmp::min(buf.len(), self.buf.len());
-        buf[..len].copy_from_slice(&self.buf[..len]);
-        self.buf.split_to(len);
-        if len == 0 {
-            if self.eof {
-                Ok(0)
-            } else {
-                Err(io::Error::new(io::ErrorKind::WouldBlock, ""))
-            }
-        } else {
-            Ok(len)
-        }
-    }
-}
-
-impl io::Write for Wrapper {
-    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
-        self.buf.extend_from_slice(buf);
-        Ok(buf.len())
-    }
-    fn flush(&mut self) -> io::Result<()> {
-        Ok(())
-    }
-}
-
-/// Payload stream with decompression support
-pub(crate) struct PayloadStream {
-    decoder: Decoder,
-    dst: BytesMut,
-}
-
-impl PayloadStream {
-    pub fn new(enc: ContentEncoding) -> PayloadStream {
-        let dec = match enc {
-            ContentEncoding::Br => Decoder::Br(
-                Box::new(BrotliDecoder::new(BytesMut::with_capacity(8192).writer()))),
-            ContentEncoding::Deflate => Decoder::Deflate(
-                Box::new(DeflateDecoder::new(BytesMut::with_capacity(8192).writer()))),
-            ContentEncoding::Gzip => Decoder::Gzip(None),
-            _ => Decoder::Identity,
-        };
-        PayloadStream{ decoder: dec, dst: BytesMut::new() }
-    }
-}
-
-impl PayloadStream {
-
-    pub fn feed_eof(&mut self) -> io::Result<Option<Bytes>> {
-        match self.decoder {
-            Decoder::Br(ref mut decoder) => {
-                match decoder.finish() {
-                    Ok(mut writer) => {
-                        let b = writer.get_mut().take().freeze();
-                        if !b.is_empty() {
-                            Ok(Some(b))
-                        } else {
-                            Ok(None)
-                        }
-                    },
-                    Err(err) => Err(err),
-                }
-            },
-            Decoder::Gzip(ref mut decoder) => {
-                if let Some(ref mut decoder) = *decoder {
-                    decoder.as_mut().get_mut().eof = true;
-
-                    loop {
-                        self.dst.reserve(8192);
-                        match decoder.read(unsafe{self.dst.bytes_mut()}) {
-                            Ok(n) =>  {
-                                if n == 0 {
-                                    return Ok(Some(self.dst.take().freeze()))
-                                } else {
-                                    unsafe{self.dst.set_len(n)};
-                                }
-                            }
-                            Err(err) => return Err(err),
-                        }
-                    }
-                } else {
-                    Ok(None)
-                }
-            },
-            Decoder::Deflate(ref mut decoder) => {
-                match decoder.try_finish() {
-                    Ok(_) => {
-                        let b = decoder.get_mut().get_mut().take().freeze();
-                        if !b.is_empty() {
-                            Ok(Some(b))
-                        } else {
-                            Ok(None)
-                        }
-                    },
-                    Err(err) => Err(err),
-                }
-            },
-            Decoder::Identity => Ok(None),
-        }
-    }
-
-    pub fn feed_data(&mut self, data: Bytes) -> io::Result<Option<Bytes>> {
-        match self.decoder {
-            Decoder::Br(ref mut decoder) => {
-                match decoder.write(&data).and_then(|_| decoder.flush()) {
-                    Ok(_) => {
-                        let b = decoder.get_mut().get_mut().take().freeze();
-                        if !b.is_empty() {
-                            Ok(Some(b))
-                        } else {
-                            Ok(None)
-                        }
-                    },
-                    Err(err) => Err(err)
-                }
-            },
-            Decoder::Gzip(ref mut decoder) => {
-                if decoder.is_none() {
-                    *decoder = Some(
-                        Box::new(GzDecoder::new(
-                            Wrapper{buf: BytesMut::from(data), eof: false})));
-                } else {
-                    let _ = decoder.as_mut().unwrap().write(&data);
-                }
-
-                loop {
-                    self.dst.reserve(8192);
-                    match decoder.as_mut().as_mut().unwrap().read(unsafe{self.dst.bytes_mut()}) {
-                        Ok(n) =>  {
-                            if n == 0 {
-                                return Ok(Some(self.dst.split_to(n).freeze()));
-                            } else {
-                                unsafe{self.dst.set_len(n)};
-                            }
-                        }
-                        Err(e) => return Err(e),
-                    }
-                }
-            },
-            Decoder::Deflate(ref mut decoder) => {
-                match decoder.write(&data).and_then(|_| decoder.flush()) {
-                    Ok(_) => {
-                        let b = decoder.get_mut().get_mut().take().freeze();
-                        if !b.is_empty() {
-                            Ok(Some(b))
-                        } else {
-                            Ok(None)
-                        }
-                    },
-                    Err(e) => Err(e),
-                }
-            },
-            Decoder::Identity => Ok(Some(data)),
-        }
-    }
-}
 
 /// Payload wrapper with content decompression support
 pub(crate) struct EncodedPayload {
@@ -357,6 +186,203 @@ impl PayloadWriter for EncodedPayload {
     }
 }
 
+pub(crate) enum Decoder {
+    Deflate(Box<DeflateDecoder<Writer>>),
+    Gzip(Option<Box<GzDecoder<Wrapper>>>),
+    Br(Box<BrotliDecoder<Writer>>),
+    Identity,
+}
+
+// should go after write::GzDecoder get implemented
+#[derive(Debug)]
+pub(crate) struct Wrapper {
+    pub buf: BytesMut,
+    pub eof: bool,
+}
+
+impl io::Read for Wrapper {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        let len = cmp::min(buf.len(), self.buf.len());
+        buf[..len].copy_from_slice(&self.buf[..len]);
+        self.buf.split_to(len);
+        if len == 0 {
+            if self.eof {
+                Ok(0)
+            } else {
+                Err(io::Error::new(io::ErrorKind::WouldBlock, ""))
+            }
+        } else {
+            Ok(len)
+        }
+    }
+}
+
+impl io::Write for Wrapper {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.buf.extend_from_slice(buf);
+        Ok(buf.len())
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+pub(crate) struct Writer {
+    buf: BytesMut,
+}
+
+impl Writer {
+    fn new() -> Writer {
+        Writer{buf: BytesMut::with_capacity(8192)}
+    }
+    fn take(&mut self) -> Bytes {
+        self.buf.take().freeze()
+    }
+}
+
+impl io::Write for Writer {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.buf.extend_from_slice(buf);
+        Ok(buf.len())
+    }
+    fn flush(&mut self) -> io::Result<()> {
+        Ok(())
+    }
+}
+
+/// Payload stream with decompression support
+pub(crate) struct PayloadStream {
+    decoder: Decoder,
+    dst: BytesMut,
+}
+
+impl PayloadStream {
+    pub fn new(enc: ContentEncoding) -> PayloadStream {
+        let dec = match enc {
+            ContentEncoding::Br => Decoder::Br(
+                Box::new(BrotliDecoder::new(Writer::new()))),
+            ContentEncoding::Deflate => Decoder::Deflate(
+                Box::new(DeflateDecoder::new(Writer::new()))),
+            ContentEncoding::Gzip => Decoder::Gzip(None),
+            _ => Decoder::Identity,
+        };
+        PayloadStream{ decoder: dec, dst: BytesMut::new() }
+    }
+}
+
+impl PayloadStream {
+
+    pub fn feed_eof(&mut self) -> io::Result<Option<Bytes>> {
+        match self.decoder {
+            Decoder::Br(ref mut decoder) => {
+                match decoder.finish() {
+                    Ok(mut writer) => {
+                        let b = writer.take();
+                        if !b.is_empty() {
+                            Ok(Some(b))
+                        } else {
+                            Ok(None)
+                        }
+                    },
+                    Err(e) => Err(e),
+                }
+            },
+            Decoder::Gzip(ref mut decoder) => {
+                if let Some(ref mut decoder) = *decoder {
+                    decoder.as_mut().get_mut().eof = true;
+
+                    loop {
+                        self.dst.reserve(8192);
+                        match decoder.read(unsafe{self.dst.bytes_mut()}) {
+                            Ok(n) =>  {
+                                if n == 0 {
+                                    return Ok(Some(self.dst.take().freeze()))
+                                } else {
+                                    unsafe{self.dst.advance_mut(n)};
+                                }
+                            }
+                            Err(e) => return Err(e),
+                        }
+                    }
+                } else {
+                    Ok(None)
+                }
+            },
+            Decoder::Deflate(ref mut decoder) => {
+                match decoder.try_finish() {
+                    Ok(_) => {
+                        let b = decoder.get_mut().take();
+                        if !b.is_empty() {
+                            Ok(Some(b))
+                        } else {
+                            Ok(None)
+                        }
+                    },
+                    Err(e) => Err(e),
+                }
+            },
+            Decoder::Identity => Ok(None),
+        }
+    }
+
+    pub fn feed_data(&mut self, data: Bytes) -> io::Result<Option<Bytes>> {
+        match self.decoder {
+            Decoder::Br(ref mut decoder) => {
+                match decoder.write(&data).and_then(|_| decoder.flush()) {
+                    Ok(_) => {
+                        let b = decoder.get_mut().take();
+                        if !b.is_empty() {
+                            Ok(Some(b))
+                        } else {
+                            Ok(None)
+                        }
+                    },
+                    Err(e) => Err(e)
+                }
+            },
+            Decoder::Gzip(ref mut decoder) => {
+                if decoder.is_none() {
+                    *decoder = Some(
+                        Box::new(GzDecoder::new(
+                            Wrapper{buf: BytesMut::from(data), eof: false})));
+                } else {
+                    let _ = decoder.as_mut().unwrap().write(&data);
+                }
+
+                loop {
+                    self.dst.reserve(8192);
+                    match decoder.as_mut().as_mut().unwrap().read(unsafe{self.dst.bytes_mut()}) {
+                        Ok(n) =>  {
+                            if n == 0 {
+                                return Ok(Some(self.dst.take().freeze()));
+                            } else {
+                                unsafe{self.dst.advance_mut(n)};
+                            }
+                        }
+                        Err(e) => {
+                            return Err(e)
+                        }
+                    }
+                }
+            },
+            Decoder::Deflate(ref mut decoder) => {
+                match decoder.write(&data).and_then(|_| decoder.flush()) {
+                    Ok(_) => {
+                        let b = decoder.get_mut().take();
+                        if !b.is_empty() {
+                            Ok(Some(b))
+                        } else {
+                            Ok(None)
+                        }
+                    },
+                    Err(e) => Err(e),
+                }
+            },
+            Decoder::Identity => Ok(Some(data)),
+        }
+    }
+}
+
 pub(crate) enum ContentEncoder {
     Deflate(DeflateEncoder<TransferEncoding>),
     Gzip(GzEncoder<TransferEncoding>),

src/server/h1.rs

@@ -10,7 +10,7 @@ use actix::Arbiter;
 use httparse;
 use http::{Uri, Method, Version, HttpTryFrom, HeaderMap};
 use http::header::{self, HeaderName, HeaderValue};
-use bytes::{Bytes, BytesMut, BufMut};
+use bytes::{Bytes, BytesMut};
 use futures::{Future, Poll, Async};
 use tokio_core::reactor::Timeout;
 
@@ -402,40 +402,49 @@ impl Reader {
         // read payload
         let done = {
             if let Some(ref mut payload) = self.payload {
-                match utils::read_from_io(io, buf) {
-                    Ok(Async::Ready(0)) => {
-                        payload.tx.set_error(PayloadError::Incomplete);
+                'buf: loop {
+                    let not_ready = match utils::read_from_io(io, buf) {
+                        Ok(Async::Ready(0)) => {
+                            payload.tx.set_error(PayloadError::Incomplete);
 
-                        // http channel should not deal with payload errors
-                        return Err(ReaderError::Payload)
-                    },
-                    Err(err) => {
-                        payload.tx.set_error(err.into());
-
-                        // http channel should not deal with payload errors
-                        return Err(ReaderError::Payload)
-                    }
-                    _ => (),
-                }
-                loop {
-                    match payload.decoder.decode(buf) {
-                        Ok(Async::Ready(Some(bytes))) => {
-                            payload.tx.feed_data(bytes);
-                            if payload.decoder.is_eof() {
-                                payload.tx.feed_eof();
-                                break true
-                            }
+                            // http channel should not deal with payload errors
+                            return Err(ReaderError::Payload)
                         },
-                        Ok(Async::Ready(None)) => {
-                            payload.tx.feed_eof();
-                            break true
-                        },
-                        Ok(Async::NotReady) =>
-                            return Ok(Async::NotReady),
+                        Ok(Async::NotReady) => true,
                         Err(err) => {
                             payload.tx.set_error(err.into());
+
+                            // http channel should not deal with payload errors
                             return Err(ReaderError::Payload)
                         }
+                        _ => false,
+                    };
+                    loop {
+                        match payload.decoder.decode(buf) {
+                            Ok(Async::Ready(Some(bytes))) => {
+                                payload.tx.feed_data(bytes);
+                                if payload.decoder.is_eof() {
+                                    payload.tx.feed_eof();
+                                    break 'buf true
+                                }
+                            },
+                            Ok(Async::Ready(None)) => {
+                                payload.tx.feed_eof();
+                                break 'buf true
+                            },
+                            Ok(Async::NotReady) => {
+                                // if buffer is full then
+                                // socket still can contain more data
+                                if not_ready {
+                                    return Ok(Async::NotReady)
+                                }
+                                continue 'buf
+                            },
+                            Err(err) => {
+                                payload.tx.set_error(err.into());
+                                return Err(ReaderError::Payload)
+                            }
+                        }
                     }
                 }
             } else {
@@ -445,16 +454,13 @@ impl Reader {
         if done { self.payload = None }
 
         // if buf is empty parse_message will always return NotReady, let's avoid that
-        let read = if buf.is_empty() {
+        if buf.is_empty() {
             match utils::read_from_io(io, buf) {
                 Ok(Async::Ready(0)) => return Err(ReaderError::Disconnect),
                 Ok(Async::Ready(_)) => (),
                 Ok(Async::NotReady) => return Ok(Async::NotReady),
                 Err(err) => return Err(ReaderError::Error(err.into()))
             }
-            false
-        } else {
-            true
         };
 
         loop {
@@ -470,22 +476,18 @@ impl Reader {
                     return Ok(Async::Ready(msg));
                 },
                 Async::NotReady => {
-                    if buf.capacity() >= MAX_BUFFER_SIZE {
+                    if buf.len() >= MAX_BUFFER_SIZE {
                         error!("MAX_BUFFER_SIZE unprocessed data reached, closing");
                         return Err(ReaderError::Error(ParseError::TooLarge));
                     }
-                    if read || buf.remaining_mut() == 0 {
-                        match utils::read_from_io(io, buf) {
-                            Ok(Async::Ready(0)) => {
-                                debug!("Ignored premature client disconnection");
-                                return Err(ReaderError::Disconnect);
-                            },
-                            Ok(Async::Ready(_)) => (),
-                            Ok(Async::NotReady) => return Ok(Async::NotReady),
-                            Err(err) => return Err(ReaderError::Error(err.into())),
-                        }
-                    } else {
-                        return Ok(Async::NotReady)
+                    match utils::read_from_io(io, buf) {
+                        Ok(Async::Ready(0)) => {
+                            debug!("Ignored premature client disconnection");
+                            return Err(ReaderError::Disconnect);
+                        },
+                        Ok(Async::Ready(_)) => (),
+                        Ok(Async::NotReady) => return Ok(Async::NotReady),
+                        Err(err) => return Err(ReaderError::Error(err.into())),
                     }
                 },
             }

src/server/settings.rs

@@ -63,7 +63,7 @@ pub(crate) struct WorkerSettings<H> {
     bytes: Rc<SharedBytesPool>,
     messages: Rc<helpers::SharedMessagePool>,
     channels: Cell<usize>,
-    node: Node<()>,
+    node: Box<Node<()>>,
 }
 
 impl<H> WorkerSettings<H> {
@@ -75,7 +75,7 @@ impl<H> WorkerSettings<H> {
             bytes: Rc::new(SharedBytesPool::new()),
             messages: Rc::new(helpers::SharedMessagePool::new()),
             channels: Cell::new(0),
-            node: Node::head(),
+            node: Box::new(Node::head()),
         }
     }
 

src/server/srv.rs

@@ -697,7 +697,7 @@ fn create_tcp_listener(addr: net::SocketAddr, backlog: i32) -> io::Result<net::T
         net::SocketAddr::V4(_) => TcpBuilder::new_v4()?,
         net::SocketAddr::V6(_) => TcpBuilder::new_v6()?,
     };
-    builder.bind(addr)?;
     builder.reuse_address(true)?;
+    builder.bind(addr)?;
     Ok(builder.listen(backlog)?)
 }

src/ws/client.rs

@@ -489,7 +489,7 @@ impl ClientWriter {
 
     /// Send text frame
     #[inline]
-    pub fn text<T: Into<String>>(&mut self, text: T) {
+    pub fn text<T: Into<Binary>>(&mut self, text: T) {
         self.write(Frame::message(text.into(), OpCode::Text, true, true));
     }
 

src/ws/context.rs

@@ -132,7 +132,7 @@ impl<A, S> WebsocketContext<A, S> where A: Actor<Context=Self> {
 
     /// Send text frame
     #[inline]
-    pub fn text<T: Into<String>>(&mut self, text: T) {
+    pub fn text<T: Into<Binary>>(&mut self, text: T) {
         self.write(Frame::message(text.into(), OpCode::Text, true, false));
     }
 

tests/test_client.rs

@@ -116,6 +116,32 @@ fn test_client_gzip_encoding() {
     assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
 }
 
+#[test]
+fn test_client_gzip_encoding_large() {
+    let data = STR.repeat(10);
+
+    let mut srv = test::TestServer::new(|app| app.handler(|req: HttpRequest| {
+        req.body()
+            .and_then(|bytes: Bytes| {
+                Ok(httpcodes::HTTPOk
+                   .build()
+                   .content_encoding(headers::ContentEncoding::Deflate)
+                   .body(bytes))
+            }).responder()}
+    ));
+
+    // client request
+    let request = srv.post()
+        .content_encoding(headers::ContentEncoding::Gzip)
+        .body(data.clone()).unwrap();
+    let response = srv.execute(request.send()).unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = srv.execute(response.body()).unwrap();
+    assert_eq!(bytes, Bytes::from(data));
+}
+
 #[test]
 fn test_client_brotli_encoding() {
     let mut srv = test::TestServer::new(|app| app.handler(|req: HttpRequest| {

tests/test_server.rs

@@ -94,6 +94,7 @@ fn test_start() {
 }
 
 #[test]
+#[cfg(unix)]
 fn test_shutdown() {
     let _ = test::TestServer::unused_addr();
     let (tx, rx) = mpsc::channel();
@@ -135,27 +136,32 @@ fn test_simple() {
 
 #[test]
 fn test_headers() {
+    let data = STR.repeat(10);
+    let srv_data = Arc::new(data.clone());
     let mut srv = test::TestServer::new(
-        |app| app.handler(|_| {
-            let mut builder = httpcodes::HTTPOk.build();
-            for idx in 0..90 {
-                builder.header(
-                    format!("X-TEST-{}", idx).as_str(),
-                    "TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
-                     TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST ");
-            }
-            builder.body(STR)}));
+        move |app| {
+            let data = srv_data.clone();
+            app.handler(move |_| {
+                let mut builder = httpcodes::HTTPOk.build();
+                for idx in 0..90 {
+                    builder.header(
+                        format!("X-TEST-{}", idx).as_str(),
+                        "TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST \
+                         TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST TEST ");
+                }
+                builder.body(data.as_ref())})
+        });
 
     let request = srv.get().finish().unwrap();
     let response = srv.execute(request.send()).unwrap();
@@ -163,7 +169,7 @@ fn test_headers() {
 
     // read response
     let bytes = srv.execute(response.body()).unwrap();
-    assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
+    assert_eq!(bytes, Bytes::from(data));
 }
 
 #[test]
@@ -202,6 +208,33 @@ fn test_body_gzip() {
     assert_eq!(Bytes::from(dec), Bytes::from_static(STR.as_ref()));
 }
 
+#[test]
+fn test_body_gzip_large() {
+    let data = STR.repeat(10);
+    let srv_data = Arc::new(data.clone());
+
+    let mut srv = test::TestServer::new(
+        move |app| {
+            let data = srv_data.clone();
+            app.handler(
+                move |_| httpcodes::HTTPOk.build()
+                    .content_encoding(headers::ContentEncoding::Gzip)
+                    .body(data.as_ref()))});
+
+    let request = srv.get().disable_decompress().finish().unwrap();
+    let response = srv.execute(request.send()).unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = srv.execute(response.body()).unwrap();
+
+    // decode
+    let mut e = GzDecoder::new(&bytes[..]);
+    let mut dec = Vec::new();
+    e.read_to_end(&mut dec).unwrap();
+    assert_eq!(Bytes::from(dec), Bytes::from(data));
+}
+
 #[test]
 fn test_body_chunked_implicit() {
     let mut srv = test::TestServer::new(
@@ -429,6 +462,35 @@ fn test_gzip_encoding() {
     assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
 }
 
+#[test]
+fn test_gzip_encoding_large() {
+    let data = STR.repeat(10);
+    let mut srv = test::TestServer::new(|app| app.handler(|req: HttpRequest| {
+        req.body()
+            .and_then(|bytes: Bytes| {
+                Ok(httpcodes::HTTPOk
+                   .build()
+                   .content_encoding(headers::ContentEncoding::Identity)
+                   .body(bytes))
+            }).responder()}
+    ));
+
+    // client request
+    let mut e = GzEncoder::new(Vec::new(), Compression::default());
+    e.write_all(data.as_ref()).unwrap();
+    let enc = e.finish().unwrap();
+
+    let request = srv.post()
+        .header(header::CONTENT_ENCODING, "gzip")
+        .body(enc.clone()).unwrap();
+    let response = srv.execute(request.send()).unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = srv.execute(response.body()).unwrap();
+    assert_eq!(bytes, Bytes::from(data));
+}
+
 #[test]
 fn test_deflate_encoding() {
     let mut srv = test::TestServer::new(|app| app.handler(|req: HttpRequest| {
@@ -457,6 +519,35 @@ fn test_deflate_encoding() {
     assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
 }
 
+#[test]
+fn test_deflate_encoding_large() {
+    let data = STR.repeat(10);
+    let mut srv = test::TestServer::new(|app| app.handler(|req: HttpRequest| {
+        req.body()
+            .and_then(|bytes: Bytes| {
+                Ok(httpcodes::HTTPOk
+                   .build()
+                   .content_encoding(headers::ContentEncoding::Identity)
+                   .body(bytes))
+            }).responder()}
+    ));
+
+    let mut e = DeflateEncoder::new(Vec::new(), Compression::default());
+    e.write_all(data.as_ref()).unwrap();
+    let enc = e.finish().unwrap();
+
+    // client request
+    let request = srv.post()
+        .header(header::CONTENT_ENCODING, "deflate")
+        .body(enc).unwrap();
+    let response = srv.execute(request.send()).unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = srv.execute(response.body()).unwrap();
+    assert_eq!(bytes, Bytes::from(data));
+}
+
 #[test]
 fn test_brotli_encoding() {
     let mut srv = test::TestServer::new(|app| app.handler(|req: HttpRequest| {
@@ -485,6 +576,35 @@ fn test_brotli_encoding() {
     assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
 }
 
+#[test]
+fn test_brotli_encoding_large() {
+    let data = STR.repeat(10);
+    let mut srv = test::TestServer::new(|app| app.handler(|req: HttpRequest| {
+        req.body()
+            .and_then(|bytes: Bytes| {
+                Ok(httpcodes::HTTPOk
+                   .build()
+                   .content_encoding(headers::ContentEncoding::Identity)
+                   .body(bytes))
+            }).responder()}
+    ));
+
+    let mut e = BrotliEncoder::new(Vec::new(), 5);
+    e.write_all(data.as_ref()).unwrap();
+    let enc = e.finish().unwrap();
+
+    // client request
+    let request = srv.post()
+        .header(header::CONTENT_ENCODING, "br")
+        .body(enc).unwrap();
+    let response = srv.execute(request.send()).unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = srv.execute(response.body()).unwrap();
+    assert_eq!(bytes, Bytes::from(data));
+}
+
 #[test]
 fn test_h2() {
     let srv = test::TestServer::new(|app| app.handler(|_|{

tools/wsload/src/wsclient.rs

@@ -19,7 +19,7 @@ use futures::Future;
 use rand::{thread_rng, Rng};
 
 use actix::prelude::*;
-use actix_web::ws::{Message, WsClientError, WsClient, WsClientWriter};
+use actix_web::ws;
 
 
 fn main() {
@@ -71,21 +71,21 @@ fn main() {
         let perf = perf_counters.clone();
         let addr = Arbiter::new(format!("test {}", t));
 
-        addr.send(actix::msgs::Execute::new(move || -> Result<(), ()> {
+        addr.do_send(actix::msgs::Execute::new(move || -> Result<(), ()> {
             let mut reps = report;
             for _ in 0..concurrency {
                 let pl2 = pl.clone();
                 let perf2 = perf.clone();
 
                 Arbiter::handle().spawn(
-                    WsClient::new(&ws).connect().unwrap()
+                    ws::Client::new(&ws).connect()
                         .map_err(|e| {
                             println!("Error: {}", e);
-                            Arbiter::system().send(actix::msgs::SystemExit(0));
+                            Arbiter::system().do_send(actix::msgs::SystemExit(0));
                             ()
                         })
                         .map(move |(reader, writer)| {
-                            let addr: SyncAddress<_> = ChatClient::create(move |ctx| {
+                            let addr: Addr<Syn, _> = ChatClient::create(move |ctx| {
                                 ChatClient::add_stream(reader, ctx);
                                 ChatClient{conn: writer,
                                            payload: pl2,
@@ -114,7 +114,7 @@ fn parse_u64_default(input: Option<&str>, default: u64) -> u64 {
 }
 
 struct ChatClient{
-    conn: WsClientWriter,
+    conn: ws::ClientWriter,
     payload: Arc<String>,
     ts: u64,
     bin: bool,
@@ -133,9 +133,9 @@ impl Actor for ChatClient {
         }
     }
 
-    fn stopping(&mut self, _: &mut Context<Self>) -> bool {
-        Arbiter::system().send(actix::msgs::SystemExit(0));
-        true
+    fn stopping(&mut self, _: &mut Context<Self>) -> Running {
+        Arbiter::system().do_send(actix::msgs::SystemExit(0));
+        Running::Stop
     }
 }
 
@@ -171,15 +171,15 @@ impl ChatClient {
 }
 
 /// Handle server websocket messages
-impl StreamHandler<Message, WsClientError> for ChatClient {
+impl StreamHandler<ws::Message, ws::ProtocolError> for ChatClient {
 
     fn finished(&mut self, ctx: &mut Context<Self>) {
         ctx.stop()
     }
 
-    fn handle(&mut self, msg: Message, ctx: &mut Context<Self>) {
+    fn handle(&mut self, msg: ws::Message, ctx: &mut Context<Self>) {
         match msg {
-            Message::Text(txt) => {
+            ws::Message::Text(txt) => {
                 if txt == self.payload.as_ref().as_str() {
                     self.perf_counters.register_request();
                     self.perf_counters.register_latency(time::precise_time_ns() - self.ts);