include payload into request

README.md

@@ -83,15 +83,14 @@ impl Actor for MyWebSocket {
 impl Route for MyWebSocket {
     type State = ();
 
-    fn request(req: &mut HttpRequest,
+    fn request(req: &mut HttpRequest, ctx: &mut HttpContext<Self>) -> RouteResult<Self>
-               payload: Payload, ctx: &mut HttpContext<Self>) -> RouteResult<Self>
     {
         // websocket handshake
         let resp = ws::handshake(req)?;
         // send HttpResponse back to peer
         ctx.start(resp);
         // convert bytes stream to a stream of `ws::Message` and handle stream
-        ctx.add_stream(ws::WsStream::new(payload));
+        ctx.add_stream(ws::WsStream::new(req));
         Reply::async(MyWebSocket)
     }
 }

examples/basic.rs

@@ -7,14 +7,14 @@ extern crate env_logger;
 extern crate futures;
 
 use actix_web::*;
-use actix_web::error::Result;
+use actix_web::error::{Error, Result};
 use actix_web::middlewares::RequestSession;
 use futures::stream::{once, Once};
 
 /// somple handle
-fn index(req: &mut HttpRequest, mut _payload: Payload, state: &()) -> Result<HttpResponse> {
+fn index(req: &mut HttpRequest, state: &()) -> Result<HttpResponse> {
     println!("{:?}", req);
-    if let Ok(ch) = _payload.readany() {
+    if let Ok(ch) = req.payload_mut().readany() {
         if let futures::Async::Ready(Some(d)) = ch {
             println!("{}", String::from_utf8_lossy(d.0.as_ref()));
         }
@@ -32,8 +32,7 @@ fn index(req: &mut HttpRequest, mut _payload: Payload, state: &()) -> Result<Htt
 }
 
 /// somple handle
-fn index_async(req: &mut HttpRequest, _payload: Payload, state: &())
+fn index_async(req: &mut HttpRequest, state: &()) -> Once<actix_web::Frame, Error>
-               -> Once<actix_web::Frame, actix_web::error::Error>
 {
     println!("{:?}", req);
 
@@ -45,7 +44,7 @@ fn index_async(req: &mut HttpRequest, _payload: Payload, state: &())
 }
 
 /// handle with path parameters like `/user/{name}/`
-fn with_param(req: &mut HttpRequest, _payload: Payload, state: &()) -> Result<HttpResponse>
+fn with_param(req: &mut HttpRequest, state: &()) -> Result<HttpResponse>
 {
     println!("{:?}", req);
 
@@ -76,7 +75,7 @@ fn main() {
             // async handler
             .resource("/async/{name}", |r| r.async(Method::GET, index_async))
             // redirect
-            .resource("/", |r| r.handler(Method::GET, |req, _, _| {
+            .resource("/", |r| r.handler(Method::GET, |req, _| {
                 println!("{:?}", req);
 
                 Ok(httpcodes::HTTPFound

examples/state.rs

@@ -15,7 +15,7 @@ struct AppState {
 }
 
 /// somple handle
-fn index(req: &mut HttpRequest, _: Payload, state: &AppState) -> HttpResponse {
+fn index(req: &mut HttpRequest, state: &AppState) -> HttpResponse {
     println!("{:?}", req);
     state.counter.set(state.counter.get() + 1);
     httpcodes::HTTPOk.with_body(
@@ -36,12 +36,11 @@ impl Route for MyWebSocket {
     /// Shared application state
     type State = AppState;
 
-    fn request(req: &mut HttpRequest,
+    fn request(req: &mut HttpRequest, ctx: &mut HttpContext<Self>) -> RouteResult<Self>
-               payload: Payload, ctx: &mut HttpContext<Self>) -> RouteResult<Self>
     {
         let resp = ws::handshake(req)?;
         ctx.start(resp);
-        ctx.add_stream(ws::WsStream::new(payload));
+        ctx.add_stream(ws::WsStream::new(req));
         Reply::async(MyWebSocket{counter: 0})
     }
 }

examples/websocket.rs

@@ -22,15 +22,14 @@ impl Actor for MyWebSocket {
 impl Route for MyWebSocket {
     type State = ();
 
-    fn request(req: &mut HttpRequest,
+    fn request(req: &mut HttpRequest, ctx: &mut HttpContext<Self>) -> RouteResult<Self>
-               payload: Payload, ctx: &mut HttpContext<Self>) -> RouteResult<Self>
     {
         // websocket handshake
         let resp = ws::handshake(req)?;
         // send HttpResponse back to peer
         ctx.start(resp);
         // convert bytes stream to a stream of `ws::Message` and register it
-        ctx.add_stream(ws::WsStream::new(payload));
+        ctx.add_stream(ws::WsStream::new(req));
         Reply::async(MyWebSocket)
     }
 }

src/application.rs

@@ -2,7 +2,6 @@ use std::rc::Rc;
 use std::collections::HashMap;
 
 use task::Task;
-use payload::Payload;
 use route::{RouteHandler, FnHandler};
 use resource::Resource;
 use recognizer::{RouteRecognizer, check_pattern};
@@ -25,19 +24,19 @@ pub struct Application<S> {
 
 impl<S: 'static> Application<S> {
 
-    fn run(&self, req: &mut HttpRequest, payload: Payload) -> Task {
+    fn run(&self, req: &mut HttpRequest) -> Task {
         if let Some((params, h)) = self.router.recognize(req.path()) {
             if let Some(params) = params {
                 req.set_match_info(params);
             }
-            h.handle(req, payload, Rc::clone(&self.state))
+            h.handle(req, Rc::clone(&self.state))
         } else {
             for (prefix, handler) in &self.handlers {
                 if req.path().starts_with(prefix) {
-                    return handler.handle(req, payload, Rc::clone(&self.state))
+                    return handler.handle(req, Rc::clone(&self.state))
                 }
             }
-            self.default.handle(req, payload, Rc::clone(&self.state))
+            self.default.handle(req, Rc::clone(&self.state))
         }
     }
 }
@@ -48,9 +47,9 @@ impl<S: 'static> HttpHandler for Application<S> {
         &self.prefix
     }
     
-    fn handle(&self, req: HttpRequest, payload: Payload) -> Pipeline {
+    fn handle(&self, req: HttpRequest) -> Pipeline {
-        Pipeline::new(req, payload, Rc::clone(&self.middlewares),
+        Pipeline::new(req, Rc::clone(&self.middlewares),
-                      &|req: &mut HttpRequest, payload: Payload| {self.run(req, payload)})
+                      &|req: &mut HttpRequest| {self.run(req)})
     }
 }
 
@@ -140,9 +139,7 @@ impl<S> ApplicationBuilder<S> where S: 'static {
     /// impl Route for MyRoute {
     ///     type State = ();
     ///
-    ///     fn request(req: &mut HttpRequest,
+    ///     fn request(req: &mut HttpRequest, ctx: &mut HttpContext<Self>) -> RouteResult<Self> {
-    ///                payload: Payload,
-    ///                ctx: &mut HttpContext<Self>) -> RouteResult<Self> {
     ///         Reply::reply(httpcodes::HTTPOk)
     ///     }
     /// }
@@ -150,7 +147,7 @@ impl<S> ApplicationBuilder<S> where S: 'static {
     ///     let app = Application::default("/")
     ///         .resource("/test", |r| {
     ///              r.get::<MyRoute>();
-    ///              r.handler(Method::HEAD, |req, payload, state| {
+    ///              r.handler(Method::HEAD, |req, state| {
     ///                  Ok(httpcodes::HTTPMethodNotAllowed)
     ///              });
     ///         })
@@ -194,7 +191,7 @@ impl<S> ApplicationBuilder<S> where S: 'static {
     ///
     /// fn main() {
     ///     let app = Application::default("/")
-    ///         .handler("/test", |req, payload, state| {
+    ///         .handler("/test", |req, state| {
     ///              match *req.method() {
     ///                  Method::GET => httpcodes::HTTPOk,
     ///                  Method::POST => httpcodes::HTTPMethodNotAllowed,
@@ -205,7 +202,7 @@ impl<S> ApplicationBuilder<S> where S: 'static {
     /// }
     /// ```
     pub fn handler<P, F, R>(&mut self, path: P, handler: F) -> &mut Self
-        where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
+        where F: Fn(&mut HttpRequest, &S) -> R + 'static,
               R: Into<HttpResponse> + 'static,
               P: Into<String>,
     {

src/channel.rs

@@ -8,7 +8,6 @@ use tokio_io::{AsyncRead, AsyncWrite};
 
 use h1;
 use h2;
-use payload::Payload;
 use pipeline::Pipeline;
 use httprequest::HttpRequest;
 
@@ -17,7 +16,7 @@ pub trait HttpHandler: 'static {
     /// Http handler prefix
     fn prefix(&self) -> &str;
     /// Handle request
-    fn handle(&self, req: HttpRequest, payload: Payload) -> Pipeline;
+    fn handle(&self, req: HttpRequest) -> Pipeline;
 }
 
 enum HttpProtocol<T, H>

src/h1.rs

@@ -167,7 +167,7 @@ impl<T, H> Http1<T, H>
             // read incoming data
             while !self.error && !self.h2 && self.tasks.len() < MAX_PIPELINED_MESSAGES {
                 match self.reader.parse(self.stream.get_mut(), &mut self.read_buf) {
-                    Ok(Async::Ready(Item::Http1(mut req, payload))) => {
+                    Ok(Async::Ready(Item::Http1(mut req))) => {
                         not_ready = false;
 
                         // set remote addr
@@ -180,7 +180,7 @@ impl<T, H> Http1<T, H>
                         let mut task = None;
                         for h in self.router.iter() {
                             if req.path().starts_with(h.prefix()) {
-                                task = Some(h.handle(req, payload));
+                                task = Some(h.handle(req));
                                 break
                             }
                         }
@@ -270,7 +270,7 @@ impl<T, H> Http1<T, H>
 
 #[derive(Debug)]
 enum Item {
-    Http1(HttpRequest, Payload),
+    Http1(HttpRequest),
     Http2,
 }
 
@@ -297,9 +297,8 @@ enum ReaderError {
     Error(ParseError),
 }
 
-#[derive(Debug)]
 enum Message {
-    Http1(HttpRequest, Option<Decoder>),
+    Http1(HttpRequest, Option<PayloadInfo>),
     Http2,
     NotReady,
 }
@@ -377,12 +376,7 @@ impl Reader {
         loop {
             match Reader::parse_message(buf).map_err(ReaderError::Error)? {
                 Message::Http1(msg, decoder) => {
-                    let payload = if let Some(decoder) = decoder {
+                    if let Some(payload) = decoder {
-                        let (tx, rx) = Payload::new(false);
-                        let payload = PayloadInfo {
-                            tx: PayloadType::new(msg.headers(), tx),
-                            decoder: decoder,
-                        };
                         self.payload = Some(payload);
 
                         loop {
@@ -419,13 +413,9 @@ impl Reader {
                                 }
                             }
                         }
-                        rx
+                    }
-                    } else {
-                        let (_, rx) = Payload::new(true);
-                        rx
-                    };
                     self.h1 = true;
-                    return Ok(Async::Ready(Item::Http1(msg, payload)));
+                    return Ok(Async::Ready(Item::Http1(msg)));
                 },
                 Message::Http2 => {
                     if self.h1 {
@@ -566,21 +556,31 @@ impl Reader {
             }
         }
 
-        let msg = HttpRequest::new(method, path, version, headers, query);
+        let (mut psender, payload) = Payload::new(false);
+        let msg = HttpRequest::new(method, path, version, headers, query, payload);
 
         let decoder = if msg.upgrade() {
-            Some(Decoder::eof())
+            Decoder::eof()
         } else {
-            let chunked = msg.chunked()?;
+            let has_len = msg.headers().contains_key(header::CONTENT_LENGTH);
 
-            // Content-Length
+            // Chunked encoding
-            if let Some(len) = msg.headers().get(header::CONTENT_LENGTH) {
+            if msg.chunked()? {
-                if chunked {
+                if has_len {
                     return Err(ParseError::Header)
                 }
+                Decoder::chunked()
+            } else {
+                if !has_len {
+                    psender.feed_eof();
+                    return Ok(Message::Http1(msg, None))
+                }
+
+                // Content-Length
+                let len = msg.headers().get(header::CONTENT_LENGTH).unwrap();
                 if let Ok(s) = len.to_str() {
                     if let Ok(len) = s.parse::<u64>() {
-                        Some(Decoder::length(len))
+                        Decoder::length(len)
                     } else {
                         debug!("illegal Content-Length: {:?}", len);
                         return Err(ParseError::Header)
@@ -589,13 +589,14 @@ impl Reader {
                     debug!("illegal Content-Length: {:?}", len);
                     return Err(ParseError::Header)
                 }
-            } else if chunked {
-                Some(Decoder::chunked())
-            } else {
-                None
             }
         };
-        Ok(Message::Http1(msg, decoder))
+
+        let payload = PayloadInfo {
+            tx: PayloadType::new(msg.headers(), psender),
+            decoder: decoder,
+        };
+        Ok(Message::Http1(msg, Some(payload)))
     }
 }
 
@@ -940,7 +941,7 @@ mod tests {
     macro_rules! parse_ready {
         ($e:expr) => (
             match Reader::new().parse($e, &mut BytesMut::new()) {
-                Ok(Async::Ready(Item::Http1(req, payload))) => (req, payload),
+                Ok(Async::Ready(Item::Http1(req))) => req,
                 Ok(_) => panic!("Eof during parsing http request"),
                 Err(err) => panic!("Error during parsing http request: {:?}", err),
             }
@@ -950,7 +951,7 @@ mod tests {
     macro_rules! reader_parse_ready {
         ($e:expr) => (
             match $e {
-                Ok(Async::Ready(Item::Http1(req, payload))) => (req, payload),
+                Ok(Async::Ready(Item::Http1(req))) => req,
                 Ok(_) => panic!("Eof during parsing http request"),
                 Err(err) => panic!("Error during parsing http request: {:?}", err),
             }
@@ -979,11 +980,11 @@ mod tests {
 
         let mut reader = Reader::new();
         match reader.parse(&mut buf, &mut readbuf) {
-            Ok(Async::Ready(Item::Http1(req, payload))) => {
+            Ok(Async::Ready(Item::Http1(req))) => {
                 assert_eq!(req.version(), Version::HTTP_11);
                 assert_eq!(*req.method(), Method::GET);
                 assert_eq!(req.path(), "/test");
-                assert!(payload.eof());
+                assert!(req.payload().eof());
             }
             Ok(_) | Err(_) => panic!("Error during parsing http request"),
         }
@@ -1002,11 +1003,11 @@ mod tests {
 
         buf.feed_data(".1\r\n\r\n");
         match reader.parse(&mut buf, &mut readbuf) {
-            Ok(Async::Ready(Item::Http1(req, payload))) => {
+            Ok(Async::Ready(Item::Http1(req))) => {
                 assert_eq!(req.version(), Version::HTTP_11);
                 assert_eq!(*req.method(), Method::PUT);
                 assert_eq!(req.path(), "/test");
-                assert!(payload.eof());
+                assert!(req.payload().eof());
             }
             Ok(_) | Err(_) => panic!("Error during parsing http request"),
         }
@@ -1019,11 +1020,11 @@ mod tests {
 
         let mut reader = Reader::new();
         match reader.parse(&mut buf, &mut readbuf) {
-            Ok(Async::Ready(Item::Http1(req, payload))) => {
+            Ok(Async::Ready(Item::Http1(req))) => {
                 assert_eq!(req.version(), Version::HTTP_10);
                 assert_eq!(*req.method(), Method::POST);
                 assert_eq!(req.path(), "/test2");
-                assert!(payload.eof());
+                assert!(req.payload().eof());
             }
             Ok(_) | Err(_) => panic!("Error during parsing http request"),
         }
@@ -1036,11 +1037,11 @@ mod tests {
 
         let mut reader = Reader::new();
         match reader.parse(&mut buf, &mut readbuf) {
-            Ok(Async::Ready(Item::Http1(req, mut payload))) => {
+            Ok(Async::Ready(Item::Http1(mut req))) => {
                 assert_eq!(req.version(), Version::HTTP_11);
                 assert_eq!(*req.method(), Method::GET);
                 assert_eq!(req.path(), "/test");
-                assert_eq!(payload.readall().unwrap().as_ref(), b"body");
+                assert_eq!(req.payload_mut().readall().unwrap().as_ref(), b"body");
             }
             Ok(_) | Err(_) => panic!("Error during parsing http request"),
         }
@@ -1054,11 +1055,11 @@ mod tests {
 
         let mut reader = Reader::new();
         match reader.parse(&mut buf, &mut readbuf) {
-            Ok(Async::Ready(Item::Http1(req, mut payload))) => {
+            Ok(Async::Ready(Item::Http1(mut req))) => {
                 assert_eq!(req.version(), Version::HTTP_11);
                 assert_eq!(*req.method(), Method::GET);
                 assert_eq!(req.path(), "/test");
-                assert_eq!(payload.readall().unwrap().as_ref(), b"body");
+                assert_eq!(req.payload_mut().readall().unwrap().as_ref(), b"body");
             }
             Ok(_) | Err(_) => panic!("Error during parsing http request"),
         }
@@ -1074,11 +1075,11 @@ mod tests {
 
         buf.feed_data("\r\n");
         match reader.parse(&mut buf, &mut readbuf) {
-            Ok(Async::Ready(Item::Http1(req, payload))) => {
+            Ok(Async::Ready(Item::Http1(req))) => {
                 assert_eq!(req.version(), Version::HTTP_11);
                 assert_eq!(*req.method(), Method::GET);
                 assert_eq!(req.path(), "/test");
-                assert!(payload.eof());
+                assert!(req.payload().eof());
             }
             Ok(_) | Err(_) => panic!("Error during parsing http request"),
         }
@@ -1100,12 +1101,12 @@ mod tests {
 
         buf.feed_data("t: value\r\n\r\n");
         match reader.parse(&mut buf, &mut readbuf) {
-            Ok(Async::Ready(Item::Http1(req, payload))) => {
+            Ok(Async::Ready(Item::Http1(req))) => {
                 assert_eq!(req.version(), Version::HTTP_11);
                 assert_eq!(*req.method(), Method::GET);
                 assert_eq!(req.path(), "/test");
                 assert_eq!(req.headers().get("test").unwrap().as_bytes(), b"value");
-                assert!(payload.eof());
+                assert!(req.payload().eof());
             }
             Ok(_) | Err(_) => panic!("Error during parsing http request"),
         }
@@ -1121,7 +1122,7 @@ mod tests {
 
         let mut reader = Reader::new();
         match reader.parse(&mut buf, &mut readbuf) {
-            Ok(Async::Ready(Item::Http1(req, _))) => {
+            Ok(Async::Ready(Item::Http1(req))) => {
                 let val: Vec<_> = req.headers().get_all("Set-Cookie")
                     .iter().map(|v| v.to_str().unwrap().to_owned()).collect();
                 assert_eq!(val[0], "c1=cookie1");
@@ -1134,7 +1135,7 @@ mod tests {
     #[test]
     fn test_conn_default_1_0() {
         let mut buf = Buffer::new("GET /test HTTP/1.0\r\n\r\n");
-        let (req, _) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert!(!req.keep_alive());
     }
@@ -1142,7 +1143,7 @@ mod tests {
     #[test]
     fn test_conn_default_1_1() {
         let mut buf = Buffer::new("GET /test HTTP/1.1\r\n\r\n");
-        let (req, _) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert!(req.keep_alive());
     }
@@ -1152,7 +1153,7 @@ mod tests {
         let mut buf = Buffer::new(
             "GET /test HTTP/1.1\r\n\
              connection: close\r\n\r\n");
-        let (req, _) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert!(!req.keep_alive());
     }
@@ -1162,7 +1163,7 @@ mod tests {
         let mut buf = Buffer::new(
             "GET /test HTTP/1.0\r\n\
              connection: close\r\n\r\n");
-        let (req, _) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert!(!req.keep_alive());
     }
@@ -1172,7 +1173,7 @@ mod tests {
         let mut buf = Buffer::new(
             "GET /test HTTP/1.0\r\n\
              connection: keep-alive\r\n\r\n");
-        let (req, _) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert!(req.keep_alive());
     }
@@ -1182,7 +1183,7 @@ mod tests {
         let mut buf = Buffer::new(
             "GET /test HTTP/1.1\r\n\
              connection: keep-alive\r\n\r\n");
-        let (req, _) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert!(req.keep_alive());
     }
@@ -1192,7 +1193,7 @@ mod tests {
         let mut buf = Buffer::new(
             "GET /test HTTP/1.0\r\n\
              connection: other\r\n\r\n");
-        let (req, _) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert!(!req.keep_alive());
     }
@@ -1202,7 +1203,7 @@ mod tests {
         let mut buf = Buffer::new(
             "GET /test HTTP/1.1\r\n\
              connection: other\r\n\r\n");
-        let (req, _) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert!(req.keep_alive());
     }
@@ -1212,9 +1213,9 @@ mod tests {
         let mut buf = Buffer::new(
             "GET /test HTTP/1.1\r\n\
              connection: upgrade\r\n\r\n");
-        let (req, payload) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
-        assert!(!payload.eof());
+        assert!(!req.payload().eof());
         assert!(req.upgrade());
     }
 
@@ -1223,10 +1224,10 @@ mod tests {
         let mut buf = Buffer::new(
             "CONNECT /test HTTP/1.1\r\n\
              content-length: 0\r\n\r\n");
-        let (req, payload) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert!(req.upgrade());
-        assert!(!payload.eof());
+        assert!(!req.payload().eof());
     }
 
     #[test]
@@ -1234,9 +1235,9 @@ mod tests {
         let mut buf = Buffer::new(
             "GET /test HTTP/1.1\r\n\
              transfer-encoding: chunked\r\n\r\n");
-        let (req, payload) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
-        assert!(!payload.eof());
+        assert!(!req.payload().eof());
         if let Ok(val) = req.chunked() {
             assert!(val);
         } else {
@@ -1246,9 +1247,9 @@ mod tests {
         let mut buf = Buffer::new(
             "GET /test HTTP/1.1\r\n\
              transfer-encoding: chnked\r\n\r\n");
-        let (req, payload) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
-        assert!(payload.eof());
+        assert!(req.payload().eof());
         if let Ok(val) = req.chunked() {
             assert!(!val);
         } else {
@@ -1305,10 +1306,10 @@ mod tests {
              connection: upgrade\r\n\
              upgrade: websocket\r\n\r\n\
              some raw data");
-        let (req, mut payload) = parse_ready!(&mut buf);
+        let mut req = parse_ready!(&mut buf);
         assert!(!req.keep_alive());
         assert!(req.upgrade());
-        assert_eq!(payload.readall().unwrap().as_ref(), b"some raw data");
+        assert_eq!(req.payload_mut().readall().unwrap().as_ref(), b"some raw data");
     }
 
     #[test]
@@ -1316,7 +1317,7 @@ mod tests {
         let mut buf = Buffer::new(
             "GET /test HTTP/1.1\r\n\
              x-test: тест\r\n\r\n");
-        let (req, _) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert_eq!(req.headers().get("x-test").unwrap().as_bytes(),
                    "тест".as_bytes());
@@ -1326,7 +1327,7 @@ mod tests {
     fn test_http_request_parser_two_slashes() {
         let mut buf = Buffer::new(
             "GET //path HTTP/1.1\r\n\r\n");
-        let (req, _) = parse_ready!(&mut buf);
+        let req = parse_ready!(&mut buf);
 
         assert_eq!(req.path(), "//path");
     }
@@ -1354,15 +1355,15 @@ mod tests {
         let mut readbuf = BytesMut::new();
 
         let mut reader = Reader::new();
-        let (req, mut payload) = reader_parse_ready!(reader.parse(&mut buf, &mut readbuf));
+        let mut req = reader_parse_ready!(reader.parse(&mut buf, &mut readbuf));
         assert!(req.chunked().unwrap());
-        assert!(!payload.eof());
+        assert!(!req.payload().eof());
 
         buf.feed_data("4\r\ndata\r\n4\r\nline\r\n0\r\n\r\n");
         not_ready!(reader.parse(&mut buf, &mut readbuf));
-        assert!(!payload.eof());
+        assert!(!req.payload().eof());
-        assert_eq!(payload.readall().unwrap().as_ref(), b"dataline");
+        assert_eq!(req.payload_mut().readall().unwrap().as_ref(), b"dataline");
-        assert!(payload.eof());
+        assert!(req.payload().eof());
     }
 
     #[test]
@@ -1374,22 +1375,22 @@ mod tests {
 
         let mut reader = Reader::new();
 
-        let (req, mut payload) = reader_parse_ready!(reader.parse(&mut buf, &mut readbuf));
+        let mut req = reader_parse_ready!(reader.parse(&mut buf, &mut readbuf));
         assert!(req.chunked().unwrap());
-        assert!(!payload.eof());
+        assert!(!req.payload().eof());
 
         buf.feed_data(
             "4\r\ndata\r\n4\r\nline\r\n0\r\n\r\n\
              POST /test2 HTTP/1.1\r\n\
              transfer-encoding: chunked\r\n\r\n");
 
-        let (req2, payload2) = reader_parse_ready!(reader.parse(&mut buf, &mut readbuf));
+        let req2 = reader_parse_ready!(reader.parse(&mut buf, &mut readbuf));
         assert_eq!(*req2.method(), Method::POST);
         assert!(req2.chunked().unwrap());
-        assert!(!payload2.eof());
+        assert!(!req2.payload().eof());
 
-        assert_eq!(payload.readall().unwrap().as_ref(), b"dataline");
+        assert_eq!(req.payload_mut().readall().unwrap().as_ref(), b"dataline");
-        assert!(payload.eof());
+        assert!(req.payload().eof());
     }
 
     #[test]
@@ -1400,9 +1401,9 @@ mod tests {
         let mut readbuf = BytesMut::new();
 
         let mut reader = Reader::new();
-        let (req, mut payload) = reader_parse_ready!(reader.parse(&mut buf, &mut readbuf));
+        let mut req = reader_parse_ready!(reader.parse(&mut buf, &mut readbuf));
         assert!(req.chunked().unwrap());
-        assert!(!payload.eof());
+        assert!(!req.payload().eof());
 
         buf.feed_data("4\r\ndata\r");
         not_ready!(reader.parse(&mut buf, &mut readbuf));
@@ -1424,12 +1425,12 @@ mod tests {
         //buf.feed_data("test: test\r\n");
         //not_ready!(reader.parse(&mut buf, &mut readbuf));
 
-        assert_eq!(payload.readall().unwrap().as_ref(), b"dataline");
+        assert_eq!(req.payload_mut().readall().unwrap().as_ref(), b"dataline");
-        assert!(!payload.eof());
+        assert!(!req.payload().eof());
 
         buf.feed_data("\r\n");
         not_ready!(reader.parse(&mut buf, &mut readbuf));
-        assert!(payload.eof());
+        assert!(req.payload().eof());
     }
 
     #[test]
@@ -1440,15 +1441,15 @@ mod tests {
         let mut readbuf = BytesMut::new();
 
         let mut reader = Reader::new();
-        let (req, mut payload) = reader_parse_ready!(reader.parse(&mut buf, &mut readbuf));
+        let mut req = reader_parse_ready!(reader.parse(&mut buf, &mut readbuf));
         assert!(req.chunked().unwrap());
-        assert!(!payload.eof());
+        assert!(!req.payload().eof());
 
         buf.feed_data("4;test\r\ndata\r\n4\r\nline\r\n0\r\n\r\n"); // test: test\r\n\r\n")
         not_ready!(reader.parse(&mut buf, &mut readbuf));
-        assert!(!payload.eof());
+        assert!(!req.payload().eof());
-        assert_eq!(payload.readall().unwrap().as_ref(), b"dataline");
+        assert_eq!(req.payload_mut().readall().unwrap().as_ref(), b"dataline");
-        assert!(payload.eof());
+        assert!(req.payload().eof());
     }
 
     /*#[test]

src/h2.rs

@@ -220,15 +220,15 @@ impl Entry {
         let path = parts.uri.path().to_owned();
         let query = parts.uri.query().unwrap_or("").to_owned();
 
+        // Payload and Content-Encoding
+        let (psender, payload) = Payload::new(false);
+
         let mut req = HttpRequest::new(
-            parts.method, path, parts.version, parts.headers, query);
+            parts.method, path, parts.version, parts.headers, query, payload);
 
         // set remote addr
         req.set_remove_addr(addr);
 
-        // Payload and Content-Encoding
-        let (psender, payload) = Payload::new(false);
-
         // Payload sender
         let psender = PayloadType::new(req.headers(), psender);
 
@@ -236,7 +236,7 @@ impl Entry {
         let mut task = None;
         for h in router.iter() {
             if req.path().starts_with(h.prefix()) {
-                task = Some(h.handle(req, payload));
+                task = Some(h.handle(req));
                 break
             }
         }

src/httpcodes.rs

@@ -6,7 +6,6 @@ use http::StatusCode;
 use body::Body;
 use task::Task;
 use route::RouteHandler;
-use payload::Payload;
 use httprequest::HttpRequest;
 use httpresponse::{HttpResponse, HttpResponseBuilder};
 
@@ -71,7 +70,7 @@ impl StaticResponse {
 }
 
 impl<S> RouteHandler<S> for StaticResponse {
-    fn handle(&self, _: &mut HttpRequest, _: Payload, _: Rc<S>) -> Task {
+    fn handle(&self, _: &mut HttpRequest, _: Rc<S>) -> Task {
         Task::reply(HttpResponse::new(self.0, Body::Empty))
     }
 }

src/httprequest.rs

@@ -1,5 +1,5 @@
 //! HTTP Request message related code.
-use std::{str, fmt};
+use std::{str, fmt, mem};
 use std::net::SocketAddr;
 use std::collections::HashMap;
 use bytes::BytesMut;
@@ -26,13 +26,14 @@ pub struct HttpRequest {
     cookies_loaded: bool,
     extensions: Extensions,
     addr: Option<SocketAddr>,
+    payload: Payload,
 }
 
 impl HttpRequest {
     /// Construct a new Request.
     #[inline]
     pub fn new(method: Method, path: String,
-               version: Version, headers: HeaderMap, query: String) -> Self
+               version: Version, headers: HeaderMap, query: String, payload: Payload) -> Self
     {
         HttpRequest {
             method: method,
@@ -45,6 +46,7 @@ impl HttpRequest {
             cookies_loaded: false,
             extensions: Extensions::new(),
             addr: None,
+            payload: payload,
         }
     }
 
@@ -60,6 +62,7 @@ impl HttpRequest {
             cookies_loaded: false,
             extensions: Extensions::new(),
             addr: None,
+            payload: Payload::empty(),
         }
     }
 
@@ -198,7 +201,7 @@ impl HttpRequest {
 
     /// Check if request requires connection upgrade
     pub(crate) fn upgrade(&self) -> bool {
-        if let Some(conn) = self.headers().get(header::CONNECTION) {
+        if let Some(conn) = self.headers.get(header::CONNECTION) {
             if let Ok(s) = conn.to_str() {
                 return s.to_lowercase().contains("upgrade")
             }
@@ -208,7 +211,7 @@ impl HttpRequest {
 
     /// Check if request has chunked transfer encoding
     pub fn chunked(&self) -> Result<bool, ParseError> {
-        if let Some(encodings) = self.headers().get(header::TRANSFER_ENCODING) {
+        if let Some(encodings) = self.headers.get(header::TRANSFER_ENCODING) {
             if let Ok(s) = encodings.to_str() {
                 Ok(s.to_lowercase().contains("chunked"))
             } else {
@@ -230,6 +233,23 @@ impl HttpRequest {
         }
     }
 
+    /// Returns reference to the associated http payload.
+    #[inline]
+    pub fn payload(&self) -> &Payload {
+        &self.payload
+    }
+
+    /// Returns mutable reference to the associated http payload.
+    #[inline]
+    pub fn payload_mut(&mut self) -> &mut Payload {
+        &mut self.payload
+    }
+
+    /// Return payload
+    pub fn take_payload(&mut self) -> Payload {
+        mem::replace(&mut self.payload, Payload::empty())
+    }
+    
     /// Return stream to process BODY as multipart.
     ///
     /// Content-type: multipart/form-data;
@@ -344,7 +364,7 @@ mod tests {
         headers.insert(header::TRANSFER_ENCODING,
                        header::HeaderValue::from_static("chunked"));
         let req = HttpRequest::new(
-            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new());
+            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new(), Payload::empty());
 
         let (_, payload) = Payload::new(false);
         assert!(req.urlencoded(payload).is_err());
@@ -355,7 +375,7 @@ mod tests {
         headers.insert(header::CONTENT_LENGTH,
                        header::HeaderValue::from_static("xxxx"));
         let req = HttpRequest::new(
-            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new());
+            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new(), Payload::empty());
 
         let (_, payload) = Payload::new(false);
         assert!(req.urlencoded(payload).is_err());
@@ -366,7 +386,7 @@ mod tests {
         headers.insert(header::CONTENT_LENGTH,
                        header::HeaderValue::from_static("1000000"));
         let req = HttpRequest::new(
-            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new());
+            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new(), Payload::empty());
 
         let (_, payload) = Payload::new(false);
         assert!(req.urlencoded(payload).is_err());
@@ -377,10 +397,9 @@ mod tests {
         headers.insert(header::CONTENT_LENGTH,
                        header::HeaderValue::from_static("10"));
         let req = HttpRequest::new(
-            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new());
+            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new(), Payload::empty());
 
         let (_, payload) = Payload::new(false);
         assert!(req.urlencoded(payload).is_err());
     }
-
 }

src/middlewares/logger.rs

@@ -291,6 +291,7 @@ mod tests {
     use time;
     use http::{Method, Version, StatusCode};
     use http::header::{self, HeaderMap};
+    use payload::Payload;
 
     #[test]
     fn test_logger() {
@@ -299,7 +300,7 @@ mod tests {
         let mut headers = HeaderMap::new();
         headers.insert(header::USER_AGENT, header::HeaderValue::from_static("ACTIX-WEB"));
         let req = HttpRequest::new(
-            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new());
+            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new(), Payload::empty());
         let resp = HttpResponse::builder(StatusCode::OK)
             .header("X-Test", "ttt")
             .force_close().body(Body::Empty).unwrap();
@@ -330,7 +331,7 @@ mod tests {
         let mut headers = HeaderMap::new();
         headers.insert(header::USER_AGENT, header::HeaderValue::from_static("ACTIX-WEB"));
         let req = HttpRequest::new(
-            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new());
+            Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new(), Payload::empty());
         let resp = HttpResponse::builder(StatusCode::OK)
             .force_close().body(Body::Empty).unwrap();
         let entry_time = time::now();
@@ -347,7 +348,8 @@ mod tests {
         assert!(s.contains("ACTIX-WEB"));
 
         let req = HttpRequest::new(
-            Method::GET, "/".to_owned(), Version::HTTP_11, HeaderMap::new(), "test".to_owned());
+            Method::GET, "/".to_owned(), Version::HTTP_11, HeaderMap::new(),
+            "test".to_owned(), Payload::empty());
         let resp = HttpResponse::builder(StatusCode::OK)
             .force_close().body(Body::Empty).unwrap();
         let entry_time = time::now();

src/payload.rs

@@ -42,6 +42,12 @@ impl Payload {
         (PayloadSender{inner: Rc::downgrade(&shared)}, Payload{inner: shared})
     }
 
+    /// Create empty payload
+    #[doc(hidden)]
+    pub fn empty() -> Payload {
+        Payload{inner: Rc::new(RefCell::new(Inner::new(true)))}
+    }
+
     /// Indicates EOF of payload
     pub fn eof(&self) -> bool {
         self.inner.borrow().eof()

src/pipeline.rs

@@ -5,14 +5,13 @@ use futures::{Async, Poll, Future};
 
 use task::Task;
 use error::Error;
-use payload::Payload;
 use middlewares::{Middleware, Finished, Started, Response};
 use h1writer::Writer;
 use httprequest::HttpRequest;
 use httpresponse::HttpResponse;
 
-type Handler = Fn(&mut HttpRequest, Payload) -> Task;
+type Handler = Fn(&mut HttpRequest) -> Task;
-pub(crate) type PipelineHandler<'a> = &'a Fn(&mut HttpRequest, Payload) -> Task;
+pub(crate) type PipelineHandler<'a> = &'a Fn(&mut HttpRequest) -> Task;
 
 pub struct Pipeline(PipelineState);
 
@@ -27,13 +26,13 @@ enum PipelineState {
 
 impl Pipeline {
 
-    pub fn new(mut req: HttpRequest, payload: Payload,
+    pub fn new(mut req: HttpRequest,
                mw: Rc<Vec<Box<Middleware>>>, handler: PipelineHandler) -> Pipeline {
         if mw.is_empty() {
-            let task = (handler)(&mut req, payload);
+            let task = (handler)(&mut req);
             Pipeline(PipelineState::Task(Box::new((task, req))))
         } else {
-            match Start::init(mw, req, handler, payload) {
+            match Start::init(mw, req, handler) {
                 Ok(StartResult::Ready(res)) =>
                     Pipeline(PipelineState::Handle(res)),
                 Ok(StartResult::NotReady(res)) =>
@@ -154,7 +153,6 @@ struct Start {
     idx: usize,
     hnd: *mut Handler,
     disconnected: bool,
-    payload: Option<Payload>,
     fut: Option<Fut>,
     middlewares: Rc<Vec<Box<Middleware>>>,
 }
@@ -167,15 +165,12 @@ enum StartResult {
 impl Start {
 
     fn init(mw: Rc<Vec<Box<Middleware>>>,
-            req: HttpRequest,
+            req: HttpRequest, handler: PipelineHandler) -> Result<StartResult, Error> {
-            handler: PipelineHandler,
-            payload: Payload) -> Result<StartResult, Error> {
         Start {
             idx: 0,
             fut: None,
             disconnected: false,
             hnd: handler as *const _ as *mut _,
-            payload: Some(payload),
             middlewares: mw,
         }.start(req)
     }
@@ -196,8 +191,7 @@ impl Start {
         let len = self.middlewares.len();
         loop {
             if self.idx == len {
-                let task = (unsafe{&*self.hnd})(
+                let task = (unsafe{&*self.hnd})(&mut req);
-                    &mut req, self.payload.take().expect("Something is completlywrong"));
                 return Ok(StartResult::Ready(
                     Box::new(Handle::new(self.idx-1, req, self.prepare(task), self.middlewares))))
             } else {
@@ -249,8 +243,7 @@ impl Start {
                             self.prepare(Task::reply(resp)), Rc::clone(&self.middlewares)))))
                     }
                     if self.idx == len {
-                        let task = (unsafe{&*self.hnd})(
+                        let task = (unsafe{&*self.hnd})(&mut req);
-                            &mut req, self.payload.take().expect("Something is completlywrong"));
                         return Ok(Async::Ready(Box::new(Handle::new(
                             self.idx-1, req, self.prepare(task), Rc::clone(&self.middlewares)))))
                     } else {

src/resource.rs

@@ -9,7 +9,6 @@ use futures::Stream;
 use task::Task;
 use error::{Result, Error};
 use route::{Route, RouteHandler, RouteResult, Frame, FnHandler, StreamHandler};
-use payload::Payload;
 use context::HttpContext;
 use httprequest::HttpRequest;
 use httpresponse::HttpResponse;
@@ -65,7 +64,7 @@ impl<S> Resource<S> where S: 'static {
 
     /// Register handler for specified method.
     pub fn handler<F, R>(&mut self, method: Method, handler: F)
-        where F: Fn(&mut HttpRequest, Payload, &S) -> Result<R> + 'static,
+        where F: Fn(&mut HttpRequest, &S) -> Result<R> + 'static,
               R: Into<HttpResponse> + 'static,
     {
         self.routes.insert(method, Box::new(FnHandler::new(handler)));
@@ -73,7 +72,7 @@ impl<S> Resource<S> where S: 'static {
 
     /// Register async handler for specified method.
     pub fn async<F, R>(&mut self, method: Method, handler: F)
-        where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
+        where F: Fn(&mut HttpRequest, &S) -> R + 'static,
               R: Stream<Item=Frame, Error=Error> + 'static,
     {
         self.routes.insert(method, Box::new(StreamHandler::new(handler)));
@@ -126,11 +125,11 @@ impl<S> Resource<S> where S: 'static {
 
 impl<S: 'static> RouteHandler<S> for Resource<S> {
 
-    fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<S>) -> Task {
+    fn handle(&self, req: &mut HttpRequest, state: Rc<S>) -> Task {
         if let Some(handler) = self.routes.get(req.method()) {
-            handler.handle(req, payload, state)
+            handler.handle(req, state)
         } else {
-            self.default.handle(req, payload, state)
+            self.default.handle(req, state)
         }
     }
 }

src/route.rs

@@ -11,7 +11,6 @@ use body::Binary;
 use error::{Error, ExpectError};
 use context::HttpContext;
 use resource::Reply;
-use payload::Payload;
 use httprequest::HttpRequest;
 use httpresponse::HttpResponse;
 
@@ -34,7 +33,7 @@ impl Frame {
 #[allow(unused_variables)]
 pub trait RouteHandler<S>: 'static {
     /// Handle request
-    fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<S>) -> Task;
+    fn handle(&self, req: &mut HttpRequest, state: Rc<S>) -> Task;
 
     /// Set route prefix
     fn set_prefix(&mut self, prefix: String) {}
@@ -81,8 +80,7 @@ pub trait Route: Actor {
     /// request/response or websocket connection.
     /// In that case `HttpContext::start` and `HttpContext::write` has to be used
     /// for writing response.
-    fn request(req: &mut HttpRequest,
+    fn request(req: &mut HttpRequest, ctx: &mut Self::Context) -> RouteResult<Self>;
-               payload: Payload, ctx: &mut Self::Context) -> RouteResult<Self>;
 
     /// This method creates `RouteFactory` for this actor.
     fn factory() -> RouteFactory<Self, Self::State> {
@@ -97,8 +95,7 @@ impl<A, S> RouteHandler<S> for RouteFactory<A, S>
     where A: Actor<Context=HttpContext<A>> + Route<State=S>,
           S: 'static
 {
-    fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<A::State>) -> Task
+    fn handle(&self, req: &mut HttpRequest, state: Rc<A::State>) -> Task {
-    {
         let mut ctx = HttpContext::new(state);
 
         // handle EXPECT header
@@ -107,7 +104,7 @@ impl<A, S> RouteHandler<S> for RouteFactory<A, S>
                 return Task::reply(resp)
             }
         }
-        match A::request(req, payload, &mut ctx) {
+        match A::request(req, &mut ctx) {
             Ok(reply) => reply.into(ctx),
             Err(err) => Task::reply(err),
         }
@@ -117,7 +114,7 @@ impl<A, S> RouteHandler<S> for RouteFactory<A, S>
 /// Fn() route handler
 pub(crate)
 struct FnHandler<S, R, F>
-    where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
+    where F: Fn(&mut HttpRequest, &S) -> R + 'static,
           R: Into<HttpResponse>,
           S: 'static,
 {
@@ -126,7 +123,7 @@ struct FnHandler<S, R, F>
 }
 
 impl<S, R, F> FnHandler<S, R, F>
-    where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
+    where F: Fn(&mut HttpRequest, &S) -> R + 'static,
           R: Into<HttpResponse> + 'static,
           S: 'static,
 {
@@ -136,20 +133,19 @@ impl<S, R, F> FnHandler<S, R, F>
 }
 
 impl<S, R, F> RouteHandler<S> for FnHandler<S, R, F>
-    where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
+    where F: Fn(&mut HttpRequest, &S) -> R + 'static,
           R: Into<HttpResponse> + 'static,
           S: 'static,
 {
-    fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<S>) -> Task
+    fn handle(&self, req: &mut HttpRequest, state: Rc<S>) -> Task {
-    {
+        Task::reply((self.f)(req, &state).into())
-        Task::reply((self.f)(req, payload, &state).into())
     }
 }
 
 /// Async route handler
 pub(crate)
 struct StreamHandler<S, R, F>
-    where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
+    where F: Fn(&mut HttpRequest, &S) -> R + 'static,
           R: Stream<Item=Frame, Error=Error> + 'static,
           S: 'static,
 {
@@ -158,7 +154,7 @@ struct StreamHandler<S, R, F>
 }
 
 impl<S, R, F> StreamHandler<S, R, F>
-    where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
+    where F: Fn(&mut HttpRequest, &S) -> R + 'static,
           R: Stream<Item=Frame, Error=Error> + 'static,
           S: 'static,
 {
@@ -168,12 +164,11 @@ impl<S, R, F> StreamHandler<S, R, F>
 }
 
 impl<S, R, F> RouteHandler<S> for StreamHandler<S, R, F>
-    where F: Fn(&mut HttpRequest, Payload, &S) -> R + 'static,
+    where F: Fn(&mut HttpRequest, &S) -> R + 'static,
           R: Stream<Item=Frame, Error=Error> + 'static,
           S: 'static,
 {
-    fn handle(&self, req: &mut HttpRequest, payload: Payload, state: Rc<S>) -> Task
+    fn handle(&self, req: &mut HttpRequest, state: Rc<S>) -> Task {
-    {
+        Task::with_stream((self.f)(req, &state))
-        Task::with_stream((self.f)(req, payload, &state))
     }
 }

src/staticfiles.rs

@@ -10,7 +10,6 @@ use std::path::PathBuf;
 
 use task::Task;
 use route::RouteHandler;
-use payload::Payload;
 use mime_guess::get_mime_type;
 use httprequest::HttpRequest;
 use httpresponse::HttpResponse;
@@ -138,7 +137,7 @@ impl<S: 'static> RouteHandler<S> for StaticFiles {
         }
     }
 
-    fn handle(&self, req: &mut HttpRequest, _: Payload, _: Rc<S>) -> Task {
+    fn handle(&self, req: &mut HttpRequest, _: Rc<S>) -> Task {
         if !self.accessible {
             Task::reply(HTTPNotFound)
         } else {

src/ws.rs

@@ -22,15 +22,14 @@
 //! impl Route for WsRoute {
 //!     type State = ();
 //!
-//!     fn request(req: &mut HttpRequest,
+//!     fn request(req: &mut HttpRequest, ctx: &mut HttpContext<Self>) -> RouteResult<Self>
-//!                payload: Payload, ctx: &mut HttpContext<Self>) -> RouteResult<Self>
 //!     {
 //!         // WebSocket handshake
 //!         let resp = ws::handshake(&req)?;
 //!         // Send handshake response to peer
 //!         ctx.start(resp);
 //!         // Map Payload into WsStream
-//!         ctx.add_stream(ws::WsStream::new(payload));
+//!         ctx.add_stream(ws::WsStream::new(req));
 //!         // Start ws messages processing
 //!         Reply::async(WsRoute)
 //!     }
@@ -177,8 +176,8 @@ pub struct WsStream {
 }
 
 impl WsStream {
-    pub fn new(rx: Payload) -> WsStream {
+    pub fn new(req: &mut HttpRequest) -> WsStream {
-        WsStream { rx: rx, buf: BytesMut::new(), closed: false, error_sent: false }
+        WsStream { rx: req.take_payload(), buf: BytesMut::new(), closed: false, error_sent: false }
     }
 }
 
@@ -330,30 +329,33 @@ impl WsWriter {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use payload::Payload;
     use http::{Method, HeaderMap, Version, header};
 
     #[test]
     fn test_handshake() {
         let req = HttpRequest::new(Method::POST, "/".to_owned(),
-                                   Version::HTTP_11, HeaderMap::new(), String::new());
+                                   Version::HTTP_11, HeaderMap::new(),
+                                   String::new(), Payload::empty());
         assert_eq!(WsHandshakeError::GetMethodRequired, handshake(&req).err().unwrap());
 
         let req = HttpRequest::new(Method::GET, "/".to_owned(),
-                                   Version::HTTP_11, HeaderMap::new(), String::new());
+                                   Version::HTTP_11, HeaderMap::new(),
+                                   String::new(), Payload::empty());
         assert_eq!(WsHandshakeError::NoWebsocketUpgrade, handshake(&req).err().unwrap());
 
         let mut headers = HeaderMap::new();
         headers.insert(header::UPGRADE,
                        header::HeaderValue::from_static("test"));
         let req = HttpRequest::new(Method::GET, "/".to_owned(),
-                                   Version::HTTP_11, headers, String::new());
+                                   Version::HTTP_11, headers, String::new(), Payload::empty());
         assert_eq!(WsHandshakeError::NoWebsocketUpgrade, handshake(&req).err().unwrap());
 
         let mut headers = HeaderMap::new();
         headers.insert(header::UPGRADE,
                        header::HeaderValue::from_static("websocket"));
         let req = HttpRequest::new(Method::GET, "/".to_owned(),
-                                   Version::HTTP_11, headers, String::new());
+                                   Version::HTTP_11, headers, String::new(), Payload::empty());
         assert_eq!(WsHandshakeError::NoConnectionUpgrade, handshake(&req).err().unwrap());
 
         let mut headers = HeaderMap::new();
@@ -362,7 +364,7 @@ mod tests {
         headers.insert(header::CONNECTION,
                        header::HeaderValue::from_static("upgrade"));
         let req = HttpRequest::new(Method::GET, "/".to_owned(),
-                                   Version::HTTP_11, headers, String::new());
+                                   Version::HTTP_11, headers, String::new(), Payload::empty());
         assert_eq!(WsHandshakeError::NoVersionHeader, handshake(&req).err().unwrap());
 
         let mut headers = HeaderMap::new();
@@ -373,7 +375,7 @@ mod tests {
         headers.insert(SEC_WEBSOCKET_VERSION,
                        header::HeaderValue::from_static("5"));
         let req = HttpRequest::new(Method::GET, "/".to_owned(),
-                                   Version::HTTP_11, headers, String::new());
+                                   Version::HTTP_11, headers, String::new(), Payload::empty());
         assert_eq!(WsHandshakeError::UnsupportedVersion, handshake(&req).err().unwrap());
 
         let mut headers = HeaderMap::new();
@@ -384,7 +386,7 @@ mod tests {
         headers.insert(SEC_WEBSOCKET_VERSION,
                        header::HeaderValue::from_static("13"));
         let req = HttpRequest::new(Method::GET, "/".to_owned(),
-                                   Version::HTTP_11, headers, String::new());
+                                   Version::HTTP_11, headers, String::new(), Payload::empty());
         assert_eq!(WsHandshakeError::BadWebsocketKey, handshake(&req).err().unwrap());
 
         let mut headers = HeaderMap::new();
@@ -397,7 +399,7 @@ mod tests {
         headers.insert(SEC_WEBSOCKET_KEY,
                        header::HeaderValue::from_static("13"));
         let req = HttpRequest::new(Method::GET, "/".to_owned(),
-                                   Version::HTTP_11, headers, String::new());
+                                   Version::HTTP_11, headers, String::new(), Payload::empty());
         assert_eq!(StatusCode::SWITCHING_PROTOCOLS, handshake(&req).unwrap().status());
     }
 }

tests/test_httprequest.rs

@@ -10,7 +10,8 @@ use http::{header, Method, Version, HeaderMap};
 #[test]
 fn test_no_request_cookies() {
     let mut req = HttpRequest::new(
-        Method::GET, "/".to_owned(), Version::HTTP_11, HeaderMap::new(), String::new());
+        Method::GET, "/".to_owned(), Version::HTTP_11, HeaderMap::new(),
+        String::new(), Payload::empty());
     assert!(req.cookies().is_empty());
     let _ = req.load_cookies();
     assert!(req.cookies().is_empty());
@@ -23,7 +24,7 @@ fn test_request_cookies() {
                    header::HeaderValue::from_static("cookie1=value1; cookie2=value2"));
 
     let mut req = HttpRequest::new(
-        Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new());
+        Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new(), Payload::empty());
     assert!(req.cookies().is_empty());
     {
         let cookies = req.load_cookies().unwrap();
@@ -47,7 +48,8 @@ fn test_request_cookies() {
 #[test]
 fn test_no_request_range_header() {
     let req = HttpRequest::new(Method::GET, "/".to_owned(),
-                               Version::HTTP_11, HeaderMap::new(), String::new());
+                               Version::HTTP_11, HeaderMap::new(),
+                               String::new(), Payload::empty());
     let ranges = req.range(100).unwrap();
     assert!(ranges.is_empty());
 }
@@ -59,7 +61,7 @@ fn test_request_range_header() {
                    header::HeaderValue::from_static("bytes=0-4"));
 
     let req = HttpRequest::new(Method::GET, "/".to_owned(),
-                               Version::HTTP_11, headers, String::new());
+                               Version::HTTP_11, headers, String::new(), Payload::empty());
     let ranges = req.range(100).unwrap();
     assert_eq!(ranges.len(), 1);
     assert_eq!(ranges[0].start, 0);
@@ -69,7 +71,8 @@ fn test_request_range_header() {
 #[test]
 fn test_request_query() {
     let req = HttpRequest::new(Method::GET, "/".to_owned(),
-                               Version::HTTP_11, HeaderMap::new(), "id=test".to_owned());
+                               Version::HTTP_11, HeaderMap::new(),
+                               "id=test".to_owned(), Payload::empty());
 
     assert_eq!(req.query_string(), "id=test");
     let query = req.query();
@@ -79,7 +82,8 @@ fn test_request_query() {
 #[test]
 fn test_request_match_info() {
     let mut req = HttpRequest::new(Method::GET, "/value/".to_owned(),
-                                   Version::HTTP_11, HeaderMap::new(), "?id=test".to_owned());
+                                   Version::HTTP_11, HeaderMap::new(),
+                                   "?id=test".to_owned(), Payload::empty());
 
     let rec = RouteRecognizer::new("/".to_owned(), vec![("/{key}/".to_owned(), 1)]);
     let (params, _) = rec.recognize(req.path()).unwrap();
@@ -92,14 +96,15 @@ fn test_request_match_info() {
 #[test]
 fn test_chunked() {
     let req = HttpRequest::new(
-        Method::GET, "/".to_owned(), Version::HTTP_11, HeaderMap::new(), String::new());
+        Method::GET, "/".to_owned(), Version::HTTP_11, HeaderMap::new(),
+        String::new(), Payload::empty());
     assert!(!req.chunked().unwrap());
 
     let mut headers = HeaderMap::new();
     headers.insert(header::TRANSFER_ENCODING,
                    header::HeaderValue::from_static("chunked"));
     let req = HttpRequest::new(
-        Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new());
+        Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new(), Payload::empty());
     assert!(req.chunked().unwrap());
 
     let mut headers = HeaderMap::new();
@@ -108,6 +113,6 @@ fn test_chunked() {
     headers.insert(header::TRANSFER_ENCODING,
                    header::HeaderValue::from_str(s).unwrap());
     let req = HttpRequest::new(
-        Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new());
+        Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new(), Payload::empty());
     assert!(req.chunked().is_err());
 }

tests/test_httpresponse.rs

@@ -14,7 +14,7 @@ fn test_response_cookies() {
                    header::HeaderValue::from_static("cookie1=value1; cookie2=value2"));
 
     let mut req = HttpRequest::new(
-        Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new());
+        Method::GET, "/".to_owned(), Version::HTTP_11, headers, String::new(), Payload::empty());
     let cookies = req.load_cookies().unwrap();
 
     let resp = httpcodes::HTTPOk

tests/test_server.rs

@@ -16,7 +16,7 @@ fn create_server<T, A>() -> HttpServer<T, A, Application<()>> {
     HttpServer::new(
         vec![Application::default("/")
              .resource("/", |r|
-                       r.handler(Method::GET, |_, _, _| {
+                       r.handler(Method::GET, |_, _| {
                            Ok(httpcodes::HTTPOk)
                        }))
              .finish()])
@@ -96,7 +96,7 @@ fn test_middlewares() {
                                             response: act_num2,
                                             finish: act_num3})
                  .resource("/", |r|
-                           r.handler(Method::GET, |_, _, _| {
+                           r.handler(Method::GET, |_, _| {
                                Ok(httpcodes::HTTPOk)
                            }))
                  .finish()])