Implement multipart with multer (#1033)

* Implement multipart with multer

* Update multer to the version where one can poll its stream

* Add a multipart example

* Add a multipart example to Cargo.toml

* Use expect when getting a part's name with multipart

* Implement error MultipartFieldMissingName for when multer does not send a part with a name field

* Reformat multipart filter with nightly

* Make MultipartFieldMissingName private

Author: jaysonsantos

Cargo.toml

@@ -27,7 +27,7 @@ hyper = { version = "0.14", features = ["stream", "server", "http1", "http2", "t
 log = "0.4"
 mime = "0.3"
 mime_guess = "2.0.0"
-multiparty = { version = "0.1", features = ["server", "futures03"], optional = true }
+multer = { version = "2.1.0", optional = true }
 scoped-tls = "1.0"
 serde = "1.0"
 serde_json = "1.0"
@@ -55,7 +55,7 @@ listenfd = "1.0"
 
 [features]
 default = ["multipart", "websocket"]
-multipart = ["multiparty"]
+multipart = ["multer"]
 websocket = ["tokio-tungstenite"]
 tls = ["tokio-rustls"]
 
@@ -97,3 +97,8 @@ required-features = ["websocket"]
 
 [[example]]
 name = "query_string"
+
+
+[[example]]
+name = "multipart"
+required-features = ["multipart"]

examples/multipart.rs

@@ -0,0 +1,28 @@
+use futures_util::TryStreamExt;
+use warp::multipart::FormData;
+use warp::Buf;
+use warp::Filter;
+
+#[tokio::main]
+async fn main() {
+    // Running curl -F [email protected] 'localhost:3030/' should print [("file", ".gitignore", "\n/target\n**/*.rs.bk\nCargo.lock\n.idea/\nwarp.iml\n")]
+    let route = warp::multipart::form().and_then(|form: FormData| async move {
+        let field_names: Vec<_> = form
+            .and_then(|mut field| async move {
+                let contents =
+                    String::from_utf8_lossy(field.data().await.unwrap().unwrap().chunk())
+                        .to_string();
+                Ok((
+                    field.name().to_string(),
+                    field.filename().unwrap().to_string(),
+                    contents,
+                ))
+            })
+            .try_collect()
+            .await
+            .unwrap();
+
+        Ok::<_, warp::Rejection>(format!("{:?}", field_names))
+    });
+    warp::serve(route).run(([127, 0, 0, 1], 3030)).await;
+}

src/filters/multipart.rs

@@ -2,6 +2,8 @@
 //!
 //! Filters that extract a multipart body for a route.
 
+use std::error::Error as StdError;
+use std::fmt::{Display, Formatter};
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -12,8 +14,7 @@ use futures_util::{future, Stream};
 use headers::ContentType;
 use hyper::Body;
 use mime::Mime;
-use multiparty::headers::Headers;
-use multiparty::server::owned_futures03::{FormData as FormDataInner, Part as PartInner};
+use multer::{Field as PartInner, Multipart as FormDataInner};
 
 use crate::filter::{Filter, FilterBase, Internal};
 use crate::reject::{self, Rejection};
@@ -33,15 +34,14 @@ pub struct FormOptions {
 ///
 /// Extracted with a `warp::multipart::form` filter.
 pub struct FormData {
-    inner: FormDataInner<BodyIoError>,
+    inner: FormDataInner<'static>,
 }
 
 /// A single "part" of a multipart/form-data body.
 ///
 /// Yielded from the `FormData` stream.
 pub struct Part {
-    headers: Headers,
-    part: PartInner<BodyIoError>,
+    part: PartInner<'static>,
 }
 
 /// Create a `Filter` to extract a `multipart/form-data` body from a request.
@@ -111,17 +111,17 @@ impl Stream for FormData {
     type Item = Result<Part, crate::Error>;
 
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
-        match Pin::new(&mut self.inner).poll_next(cx) {
+        match self.inner.poll_next_field(cx) {
             Poll::Pending => Poll::Pending,
-            Poll::Ready(Some(Ok(part))) => {
-                let headers = match part.raw_headers().parse() {
-                    Ok(headers) => headers,
-                    Err(err) => return Poll::Ready(Some(Err(crate::Error::new(err)))),
-                };
-                Poll::Ready(Some(Ok(Part { part, headers })))
+            Poll::Ready(Ok(Some(part))) => {
+                if part.name().is_some() {
+                    Poll::Ready(Some(Ok(Part { part })))
+                } else {
+                    Poll::Ready(Some(Err(crate::Error::new(MultipartFieldMissingName))))
+                }
             }
-            Poll::Ready(None) => Poll::Ready(None),
-            Poll::Ready(Some(Err(err))) => Poll::Ready(Some(Err(crate::Error::new(err)))),
+            Poll::Ready(Ok(None)) => Poll::Ready(None),
+            Poll::Ready(Err(err)) => Poll::Ready(Some(Err(crate::Error::new(err)))),
         }
     }
 }
@@ -131,17 +131,18 @@ impl Stream for FormData {
 impl Part {
     /// Get the name of this part.
     pub fn name(&self) -> &str {
-        &self.headers.name
+        self.part.name().expect("checked for name previously")
     }
 
     /// Get the filename of this part, if present.
     pub fn filename(&self) -> Option<&str> {
-        self.headers.filename.as_deref()
+        self.part.file_name()
     }
 
     /// Get the content-type of this part, if present.
     pub fn content_type(&self) -> Option<&str> {
-        self.headers.content_type.as_deref()
+        let content_type = self.part.content_type();
+        content_type.map(|t| t.type_().as_str())
     }
 
     /// Asynchronously get some of the data for this `Part`.
@@ -167,13 +168,13 @@ impl Part {
 impl fmt::Debug for Part {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         let mut builder = f.debug_struct("Part");
-        builder.field("name", &self.headers.name);
+        builder.field("name", &self.part.name());
 
-        if let Some(ref filename) = self.headers.filename {
+        if let Some(ref filename) = self.part.file_name() {
             builder.field("filename", filename);
         }
 
-        if let Some(ref mime) = self.headers.content_type {
+        if let Some(ref mime) = self.part.content_type() {
             builder.field("content_type", mime);
         }
 
@@ -207,3 +208,15 @@ impl Stream for BodyIoError {
         }
     }
 }
+
+/// An error used when a multipart field is missing a name.
+#[derive(Debug)]
+struct MultipartFieldMissingName;
+
+impl Display for MultipartFieldMissingName {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        write!(f, "Multipart field is missing a name")
+    }
+}
+
+impl StdError for MultipartFieldMissingName {}