Merge branch 'master' into patch-1

actix-files/src/lib.rs

@@ -274,10 +274,14 @@ impl Files {
     /// By default pool with 5x threads of available cpus is used.
     /// Pool size can be changed by setting ACTIX_CPU_POOL environment variable.
     pub fn new<T: Into<PathBuf>>(path: &str, dir: T) -> Files {
-        let dir = dir.into().canonicalize().unwrap_or_else(|_| PathBuf::new());
-        if !dir.is_dir() {
-            log::error!("Specified path is not a directory: {:?}", dir);
+        let orig_dir = dir.into();
+        let dir = match orig_dir.canonicalize() {
+            Ok(canon_dir) => canon_dir,
+            Err(_) => {
+                log::error!("Specified path is not a directory: {:?}", orig_dir);
+                PathBuf::new()
             }
+        };
 
         Files {
             path: path.to_string(),

actix-http/src/error.rs

@@ -109,7 +109,7 @@ impl fmt::Display for Error {
 
 impl fmt::Debug for Error {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        writeln!(f, "{:?}", &self.cause)
+        write!(f, "{:?}", &self.cause)
     }
 }
 

actix-multipart/CHANGES.md

@@ -1,5 +1,9 @@
 # Changes
 
+## [2.0.0-alpha.4] - 2019-12-xx
+
+* Multipart handling now handles Pending during read of boundary #1205
+
 ## [0.2.0-alpha.2] - 2019-12-03
 
 * Migrate to `std::future`

actix-multipart/src/server.rs

@@ -610,7 +610,7 @@ impl InnerField {
             }
 
             match payload.readline() {
-                Ok(None) => Poll::Ready(None),
+                Ok(None) => Poll::Pending,
                 Ok(Some(line)) => {
                     if line.as_ref() != b"\r\n" {
                         log::warn!("multipart field did not read all the data or it is malformed");
@@ -867,6 +867,42 @@ mod tests {
 
         (tx, rx.map(|res| res.map_err(|_| panic!())))
     }
+    // Stream that returns from a Bytes, one char at a time and Pending every other poll()
+    struct SlowStream {
+        bytes: Bytes,
+        pos: usize,
+        ready: bool,
+    }
+
+    impl SlowStream {
+        fn new(bytes: Bytes) -> SlowStream {
+            return SlowStream {
+                bytes: bytes,
+                pos: 0,
+                ready: false,
+            }
+        }
+    }
+
+    impl Stream for SlowStream {
+        type Item = Result<Bytes, PayloadError>;
+
+        fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+            let this = self.get_mut();
+            if !this.ready {
+                this.ready = true;
+                cx.waker().wake_by_ref();
+                return Poll::Pending;
+            }
+            if this.pos == this.bytes.len() {
+                return Poll::Ready(None);
+            }
+            let res = Poll::Ready(Some(Ok(this.bytes.slice(this.pos..(this.pos + 1)))));
+            this.pos += 1;
+            this.ready = false;
+            res
+        }
+    }
 
     fn create_simple_request_with_header() -> (Bytes, HeaderMap) {
         let bytes = Bytes::from(
@@ -969,12 +1005,22 @@ mod tests {
         }
     }
 
+    // Loops, collecting all bytes until end-of-field
+    async fn get_whole_field(field: &mut Field) -> BytesMut {
+        let mut b = BytesMut::new();
+        loop {
+            match field.next().await {
+                Some(Ok(chunk)) => b.extend_from_slice(&chunk),
+                None => return b,
+                _ => unreachable!(),
+            }
+        }
+    }
+
     #[actix_rt::test]
     async fn test_stream() {
-        let (sender, payload) = create_stream();
         let (bytes, headers) = create_simple_request_with_header();
-
-        sender.send(Ok(bytes)).unwrap();
+        let payload = SlowStream::new(bytes);
 
         let mut multipart = Multipart::new(&headers, payload);
         match multipart.next().await.unwrap() {
@@ -986,14 +1032,7 @@ mod tests {
                 assert_eq!(field.content_type().type_(), mime::TEXT);
                 assert_eq!(field.content_type().subtype(), mime::PLAIN);
 
-                match field.next().await.unwrap() {
-                    Ok(chunk) => assert_eq!(chunk, "test"),
-                    _ => unreachable!(),
-                }
-                match field.next().await {
-                    None => (),
-                    _ => unreachable!(),
-                }
+                assert_eq!(get_whole_field(&mut field).await, "test");
             }
             _ => unreachable!(),
         }
@@ -1003,14 +1042,7 @@ mod tests {
                 assert_eq!(field.content_type().type_(), mime::TEXT);
                 assert_eq!(field.content_type().subtype(), mime::PLAIN);
 
-                match field.next().await {
-                    Some(Ok(chunk)) => assert_eq!(chunk, "data"),
-                    _ => unreachable!(),
-                }
-                match field.next().await {
-                    None => (),
-                    _ => unreachable!(),
-                }
+                assert_eq!(get_whole_field(&mut field).await, "data");
             }
             _ => unreachable!(),
         }