refactor multipart stream creation

examples/basic.rs

@@ -95,6 +95,6 @@ fn main() {
         .bind("127.0.0.1:8080").unwrap()
         .start();
 
-    println!("Started http server: 127.0.0.1:8080");
+    println!("Starting http server: 127.0.0.1:8080");
     let _ = sys.run();
 }

examples/multipart/Cargo.toml

@@ -9,5 +9,7 @@ path = "src/main.rs"
 
 [dependencies]
 env_logger = "*"
+futures = "0.1"
 actix = "^0.3.1"
-actix-web = { git = "https://github.com/actix/actix-web.git" }
+#actix-web = { git = "https://github.com/actix/actix-web.git" }
+actix-web = { path = "../../" }

examples/multipart/client.py

@@ -2,25 +2,25 @@ import asyncio
 import aiohttp
 
 
-def req1():
+async def req1():
     with aiohttp.MultipartWriter() as writer:
         writer.append('test')
         writer.append_json({'passed': True})
 
-    resp = yield from aiohttp.request(
+    resp = await aiohttp.ClientSession().request(
         "post", 'http://localhost:8080/multipart',
         data=writer, headers=writer.headers)
     print(resp)
     assert 200 == resp.status
 
 
-def req2():
+async def req2():
     with aiohttp.MultipartWriter() as writer:
         writer.append('test')
         writer.append_json({'passed': True})
         writer.append(open('src/main.rs'))
 
-    resp = yield from aiohttp.request(
+    resp = await aiohttp.ClientSession().request(
         "post", 'http://localhost:8080/multipart',
         data=writer, headers=writer.headers)
     print(resp)

examples/multipart/src/main.rs

@@ -2,76 +2,60 @@
 extern crate actix;
 extern crate actix_web;
 extern crate env_logger;
+extern crate futures;
 
-use actix::*;
 use actix_web::*;
+use futures::{Future, Stream};
+use futures::future::{result, Either};
 
-struct MyRoute;
 
-impl Actor for MyRoute {
+fn index(mut req: HttpRequest) -> Box<Future<Item=HttpResponse, Error=Error>>
-    type Context = HttpContext<Self>;
+{
-}
+    println!("{:?}", req);
 
-impl Route for MyRoute {
+    // get multipart stream and iterate over multipart items
-    type State = ();
+    Box::new(
-
+        req.multipart()
-    fn request(mut req: HttpRequest, ctx: &mut HttpContext<Self>) -> RouteResult<Self> {
+            .map_err(Error::from)
-        println!("{:?}", req);
+            .and_then(|item| {
-
-        let multipart = req.multipart()?;
-
-        // get Multipart stream
-        WrapStream::<MyRoute>::actstream(multipart)
-            .and_then(|item, act, ctx| {
                 // Multipart stream is a stream of Fields and nested Multiparts
                 match item {
                     multipart::MultipartItem::Field(field) => {
                         println!("==== FIELD ==== {:?}", field);
 
                         // Read field's stream
-                        fut::Either::A(
+                        Either::A(
-                            field.actstream()
+                            field.map_err(Error::from)
-                                .map(|chunk, act, ctx| {
+                                .map(|chunk| {
-                                    println!(
+                                    println!("-- CHUNK: \n{}",
-                                        "-- CHUNK: \n{}",
+                                             std::str::from_utf8(&chunk.0).unwrap());})
-                                        std::str::from_utf8(&chunk.0).unwrap());
+                                .fold((), |_, _| result::<_, Error>(Ok(()))))
-                                })
-                                .finish())
                     },
                     multipart::MultipartItem::Nested(mp) => {
                         // Do nothing for nested multipart stream
-                        fut::Either::B(fut::ok(()))
+                        Either::B(result(Ok(())))
                     }
                 }
             })
             // wait until stream finish
-            .finish()
+            .fold((), |_, _| result::<_, Error>(Ok(())))
-            .map_err(|e, act, ctx| {
+            .map(|_| httpcodes::HTTPOk.response())
-                ctx.start(httpcodes::HTTPBadRequest);
+    )
-                ctx.write_eof();
-            })
-            .map(|_, act, ctx| {
-                ctx.start(httpcodes::HTTPOk);
-                ctx.write_eof();
-            })
-            .spawn(ctx);
-
-        Reply::async(MyRoute)
-    }
 }
 
 fn main() {
+    ::std::env::set_var("RUST_LOG", "actix_web=info");
     let _ = env_logger::init();
     let sys = actix::System::new("multipart-example");
 
     HttpServer::new(
-        vec![
+        || Application::new()
-            Application::default("/")
+            // enable logger
-                .resource("/multipart", |r| {
+            .middleware(middlewares::Logger::default())
-                    r.post::<MyRoute>();
+            .resource("/multipart", |r| r.method(Method::POST).a(index)))
-                }).finish()
+        .bind("127.0.0.1:8080").unwrap()
-        ])
+        .start();
-        .serve::<_, ()>("127.0.0.1:8080").unwrap();
 
+    println!("Starting http server: 127.0.0.1:8080");
     let _ = sys.run();
 }

guide/src/qs_7.md

@@ -138,7 +138,7 @@ future object that resolve to Bytes object.
 * *readuntil* method returns *Future* that resolves when specified bytes string
   matches in input bytes stream
 
-Here is example that reads request payload and prints it.
+In this example handle reads request payload chunk by chunk and prints every chunk.
 
 ```rust
 # extern crate actix_web;

src/httprequest.rs

@@ -17,7 +17,7 @@ use payload::Payload;
 use multipart::Multipart;
 use helpers::SharedHttpMessage;
 use error::{ParseError, PayloadError, UrlGenerationError,
-            MultipartError, CookieParseError, HttpRangeError, UrlencodedError};
+            CookieParseError, HttpRangeError, UrlencodedError};
 
 
 pub struct HttpMessage {
@@ -406,9 +406,8 @@ impl<S> HttpRequest<S> {
     /// Return stream to process BODY as multipart.
     ///
     /// Content-type: multipart/form-data;
-    pub fn multipart(&mut self) -> Result<Multipart, MultipartError> {
+    pub fn multipart(&mut self) -> Multipart {
-        let boundary = Multipart::boundary(self.headers())?;
+        Multipart::from_request(self)
-        Ok(Multipart::new(boundary, self.payload().clone()))
     }
 
     /// Parse `application/x-www-form-urlencoded` encoded body.

src/multipart.rs

@@ -14,6 +14,7 @@ use futures::task::{Task, current as current_task};
 
 use error::{ParseError, PayloadError, MultipartError};
 use payload::Payload;
+use httprequest::HttpRequest;
 
 const MAX_HEADERS: usize = 32;
 
@@ -26,7 +27,8 @@ const MAX_HEADERS: usize = 32;
 #[derive(Debug)]
 pub struct Multipart {
     safety: Safety,
-    inner: Rc<RefCell<InnerMultipart>>,
+    error: Option<MultipartError>,
+    inner: Option<Rc<RefCell<InnerMultipart>>>,
 }
 
 ///
@@ -66,17 +68,32 @@ struct InnerMultipart {
 }
 
 impl Multipart {
+
     /// Create multipart instance for boundary.
     pub fn new(boundary: String, payload: Payload) -> Multipart {
         Multipart {
+            error: None,
             safety: Safety::new(),
-            inner: Rc::new(RefCell::new(
+            inner: Some(Rc::new(RefCell::new(
                 InnerMultipart {
                     payload: PayloadRef::new(payload),
                     boundary: boundary,
                     state: InnerState::FirstBoundary,
                     item: InnerMultipartItem::None,
-                }))
+                })))
+        }
+    }
+
+    /// Create multipart instance for request.
+    pub fn from_request<S>(req: &mut HttpRequest<S>) -> Multipart {
+        match Multipart::boundary(req.headers()) {
+            Ok(boundary) => Multipart::new(boundary, req.payload().clone()),
+            Err(err) =>
+                Multipart {
+                    error: Some(err),
+                    safety: Safety::new(),
+                    inner: None,
+                }
         }
     }
 
@@ -107,8 +124,10 @@ impl Stream for Multipart {
     type Error = MultipartError;
 
     fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-        if self.safety.current() {
+        if let Some(err) = self.error.take() {
-            self.inner.borrow_mut().poll(&self.safety)
+            Err(err)
+        } else if self.safety.current() {
+            self.inner.as_mut().unwrap().borrow_mut().poll(&self.safety)
         } else {
             Ok(Async::NotReady)
         }
@@ -327,7 +346,9 @@ impl InnerMultipart {
 
                 Ok(Async::Ready(Some(
                     MultipartItem::Nested(
-                        Multipart{safety: safety.clone(), inner: inner}))))
+                        Multipart{safety: safety.clone(),
+                                  error: None,
+                                  inner: Some(inner)}))))
             } else {
                 let field = Rc::new(RefCell::new(InnerField::new(
                     self.payload.clone(), self.boundary.clone(), &headers)?));