Add test comparing against ferristseng multipart, add content type getter

Cargo.toml

@@ -14,4 +14,5 @@ tokio = { version = "1", default-features = false, features = [ "io-util" ] }
 tokio-util = { version = "0.7", default-features = false, features = ["io"] }
 
 [dev-dependencies]
+common-multipart-rfc7578 = "0.6.0"
 tokio = { version = "1", features = ["full"] }

src/internal.rs

@@ -157,6 +157,14 @@ impl<'a> AsyncRead for UnsendRead<'a> {
     }
 }
 
+fn poll_return(any_written: bool) -> Poll<std::io::Result<()>> {
+    if any_written {
+        Poll::Ready(Ok(()))
+    } else {
+        Poll::Pending
+    }
+}
+
 impl<R> AsyncRead for Body<R>
 where
     R: AsyncRead + Unpin,
@@ -196,39 +204,33 @@ where
                         write_crlf(buf);
                     } else {
                         self.write_clrf_to_pending();
-                        cx.waker().wake_by_ref();
                     }
-                    Poll::Ready(Ok(()))
+
+                    poll_return(buf.filled().len() > initial_len)
                 }
                 Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
                 Poll::Ready(otherwise) => Poll::Ready(otherwise),
-                Poll::Pending if buf.filled().len() > initial_len => Poll::Ready(Ok(())),
+                Poll::Pending => poll_return(buf.filled().len() > initial_len),
-                Poll::Pending => Poll::Pending,
             }
         } else if let Some(part) = self.rest.pop_front() {
-            let fill_buf = if boundary_len(&self.boundary) < buf.remaining() {
+            let mut keep_writing = true;
+
+            if keep_writing && boundary_len(&self.boundary) < buf.remaining() {
                 write_boundary(&self.boundary, buf);
-                true
             } else {
                 self.write_boundary_to_pending();
-                false
+                keep_writing = false;
             };
 
-            let fill_buf = if fill_buf && headers_len(&part) < buf.remaining() {
+            if keep_writing && headers_len(&part) < buf.remaining() {
                 write_headers(&part, buf);
-                true
             } else {
                 self.write_headers_to_pending(&part);
-                false
             };
 
             self.current = Some(part.reader);
 
-            if !fill_buf {
+            poll_return(buf.filled().len() > initial_len)
-                cx.waker().wake_by_ref();
-            }
-
-            Poll::Ready(Ok(()))
         } else if buf.remaining() > final_boundary_len(&self.boundary) {
             self.closed = true;
 
@@ -238,15 +240,13 @@ where
                 write_crlf(buf);
             } else {
                 self.write_clrf_to_pending();
-                cx.waker().wake_by_ref();
             }
-            Poll::Ready(Ok(()))
+            poll_return(buf.filled().len() > initial_len)
         } else {
             self.closed = true;
             self.write_final_boundary_to_pending();
             self.write_clrf_to_pending();
-            cx.waker().wake_by_ref();
+            poll_return(buf.filled().len() > initial_len)
-            Poll::Ready(Ok(()))
         }
     }
 }

src/lib.rs

@@ -12,6 +12,7 @@ use tokio_util::io::ReaderStream;
 
 pub struct Body<R> {
     stream: ReaderStream<internal::Body<R>>,
+    boundary: String,
 }
 
 pub struct BodyBuilder<R> {
@@ -42,11 +43,17 @@ impl<'a> Body<SendRead<'a>> {
             parts: Vec::new(),
         }
     }
+
+    pub fn content_type(&self) -> mime::Mime {
+        format!("multipart/form-data; boundary={}", self.boundary)
+            .parse()
+            .expect("Valid mime for content_type")
+    }
 }
 
 impl<'a> BodyBuilder<SendRead<'a>> {
-    pub fn boundary(mut self, boundary: String) -> Self {
+    pub fn boundary<S: Into<String>>(mut self, boundary: S) -> Self {
-        self.boundary = boundary;
+        self.boundary = boundary.into();
         self
     }
 
@@ -78,14 +85,15 @@ impl<'a> BodyBuilder<SendRead<'a>> {
             .collect();
 
         Body {
-            stream: ReaderStream::new(internal::Body::new(self.boundary, parts)),
+            stream: ReaderStream::new(internal::Body::new(self.boundary.clone(), parts)),
+            boundary: self.boundary,
         }
     }
 }
 
 impl<'a> BodyBuilder<UnsendRead<'a>> {
-    pub fn boundary(mut self, boundary: String) -> Self {
+    pub fn boundary<S: Into<String>>(mut self, boundary: S) -> Self {
-        self.boundary = boundary;
+        self.boundary = boundary.into();
         self
     }
 
@@ -107,7 +115,8 @@ impl<'a> BodyBuilder<UnsendRead<'a>> {
             .collect();
 
         Body {
-            stream: ReaderStream::new(internal::Body::new(self.boundary, parts)),
+            stream: ReaderStream::new(internal::Body::new(self.boundary.clone(), parts)),
+            boundary: self.boundary,
         }
     }
 }
@@ -117,30 +126,33 @@ fn encode(value: String) -> String {
 }
 
 impl<'a> Part<SendRead<'a>> {
-    pub fn new<R: AsyncRead + Send + 'a>(name: String, reader: R) -> Self {
+    pub fn new<S: Into<String>, R: AsyncRead + Send + 'a>(name: S, reader: R) -> Self {
         Part {
-            name,
+            name: name.into(),
             content_type: None,
             filename: None,
             reader: SendRead(Box::pin(reader)),
         }
     }
 
-    pub fn new_unsend<R: AsyncRead + 'a>(name: String, reader: R) -> Part<UnsendRead<'a>> {
+    pub fn new_unsend<S: Into<String>, R: AsyncRead + 'a>(
+        name: S,
+        reader: R,
+    ) -> Part<UnsendRead<'a>> {
         Part {
-            name,
+            name: name.into(),
             content_type: None,
             filename: None,
             reader: UnsendRead(Box::pin(reader)),
         }
     }
 
-    pub fn new_str(name: String, text: &'a str) -> Self {
+    pub fn new_str<S: Into<String>>(name: S, text: &'a str) -> Self {
         Self::new(name, text.as_bytes()).content_type(mime::TEXT_PLAIN)
     }
 
-    pub fn new_string(name: String, text: String) -> Self {
+    pub fn new_string<S1: Into<String>, S2: Into<String>>(name: S1, text: S2) -> Self {
-        Self::new(name, Cursor::new(text)).content_type(mime::TEXT_PLAIN)
+        Self::new(name, Cursor::new(text.into())).content_type(mime::TEXT_PLAIN)
     }
 
     pub fn content_type(mut self, content_type: mime::Mime) -> Self {
@@ -148,8 +160,8 @@ impl<'a> Part<SendRead<'a>> {
         self
     }
 
-    pub fn filename(mut self, filename: String) -> Self {
+    pub fn filename<S: Into<String>>(mut self, filename: S) -> Self {
-        self.filename = Some(filename);
+        self.filename = Some(filename.into());
         self
     }
 
@@ -178,8 +190,8 @@ impl<'a> Part<UnsendRead<'a>> {
         self
     }
 
-    pub fn filename(mut self, filename: String) -> Self {
+    pub fn filename<S: Into<String>>(mut self, filename: S) -> Self {
-        self.filename = Some(filename);
+        self.filename = Some(filename.into());
         self
     }
 
@@ -279,17 +291,14 @@ mod tests {
 
         let body = super::Body::builder()
             .boundary(String::from("hello"))
-            .append(super::Part::new_str(String::from("name1"), "value1"))
+            .append(super::Part::new_str("name1", "value1"))
-            .append(super::Part::new_str(String::from("name2"), "value2"))
+            .append(super::Part::new_str("name2", "value2"))
             .append(
-                super::Part::new(String::from("images[]"), file)
+                super::Part::new("images[]", file)
-                    .filename(String::from("Shenzi.webp"))
+                    .filename("Shenzi.webp")
                     .content_type("image/webp".parse().expect("Parsed mime")),
             )
-            .append(super::Part::new(
+            .append(super::Part::new("input", Cursor::new("Hello World!")))
-                String::from("input"),
-                Cursor::new("Hello World!"),
-            ))
             .build();
 
         let mut out = Vec::new();
@@ -332,10 +341,66 @@ mod tests {
         .collect::<Vec<u8>>();
 
         if out != expected {
-            tokio::fs::write("./actual.multipart", out)
+            tokio::fs::write("./simple.actual.multipart", out)
                 .await
                 .expect("Wrote file");
-            tokio::fs::write("./expected.multipart", expected)
+            tokio::fs::write("./simple.expected.multipart", expected)
+                .await
+                .expect("Wrote file");
+            panic!("No match")
+        }
+    }
+
+    struct MyBoundary;
+    impl common_multipart_rfc7578::client::multipart::BoundaryGenerator for MyBoundary {
+        fn generate_boundary() -> String {
+            String::from("MYBOUNDARY")
+        }
+    }
+
+    #[tokio::test]
+    async fn compare_feristseng() {
+        let mut form = common_multipart_rfc7578::client::multipart::Form::new::<MyBoundary>();
+        let path1 = "./files/Shenzi.webp";
+        let path2 = "./files/Shenzi.png";
+        form.add_text("meowdy", "y'all");
+        form.add_reader("images[]", std::fs::File::open(path1).expect("Opened file"));
+        form.add_reader("images[]", std::fs::File::open(path2).expect("Opened file"));
+        form.add_text("howdy", "y'all");
+
+        let body = super::Body::builder()
+            .boundary("MYBOUNDARY")
+            .append(super::Part::new_str("meowdy", "y'all"))
+            .append(super::Part::new(
+                "images[]",
+                tokio::fs::File::open(path1).await.expect("Opened file"),
+            ))
+            .append(super::Part::new(
+                "images[]",
+                tokio::fs::File::open(path2).await.expect("Opened file"),
+            ))
+            .append(super::Part::new_str("howdy", "y'all"))
+            .build();
+
+        let mut ferrisbody: Streamer<common_multipart_rfc7578::client::multipart::Body> =
+            Streamer(form.into());
+        let mut mybody = Streamer(body);
+
+        let mut ferriststeng = Vec::new();
+        let mut mine = Vec::new();
+
+        while let Some(res) = ferrisbody.next().await {
+            ferriststeng.extend(res.expect("read success"));
+        }
+        while let Some(res) = mybody.next().await {
+            mine.extend(res.expect("read success"));
+        }
+
+        if mine != ferriststeng {
+            tokio::fs::write("./compare.actual.multipart", mine)
+                .await
+                .expect("Wrote file");
+            tokio::fs::write("./compare.expected.multipart", ferriststeng)
                 .await
                 .expect("Wrote file");
             panic!("No match")