Merge pull request #75 from rmja/chunked-reader-fix

Fix bug when calling fill_buf() when there are no remaining bytes

src/response/chunked.rs

@@ -220,12 +220,13 @@ where
 {
     async fn fill_buf(&mut self) -> Result<&[u8], Self::Error> {
         let remaining = self.handle_chunk_boundary().await?;
-
-        let buf = self.raw_body.fill_buf().await.map_err(|e| Error::Network(e.kind()))?;
-
-        let len = buf.len().min(remaining);
-
-        Ok(&buf[..len])
+        if remaining == 0 {
+            Ok(&[])
+        } else {
+            let buf = self.raw_body.fill_buf().await.map_err(|e| Error::Network(e.kind()))?;
+            let len = buf.len().min(remaining);
+            Ok(&buf[..len])
+        }
     }
 
     fn consume(&mut self, amt: usize) {

tests/client.rs

@@ -1,12 +1,12 @@
-use embedded_io_async::BufRead;
+use embedded_io_async::{BufRead, Write};
 use hyper::server::conn::Http;
 use hyper::service::{make_service_fn, service_fn};
 use hyper::{Body, Server};
 use rand::rngs::OsRng;
 use rand::RngCore;
 use reqwless::client::HttpClient;
 use reqwless::headers::ContentType;
-use reqwless::request::{Method, RequestBuilder};
+use reqwless::request::{Method, RequestBody, RequestBuilder};
 use reqwless::response::Status;
 use std::net::SocketAddr;
 use std::sync::Once;
@@ -296,6 +296,59 @@ async fn test_request_response_notls_buffered() {
     t.await.unwrap();
 }
 
+struct ChunkedBody(&'static [&'static [u8]]);
+
+impl RequestBody for ChunkedBody {
+    fn len(&self) -> Option<usize> {
+        None // Unknown length: triggers chunked body
+    }
+
+    async fn write<W: Write>(&self, writer: &mut W) -> Result<(), W::Error> {
+        for chunk in self.0 {
+            writer.write_all(chunk).await?;
+        }
+        Ok(())
+    }
+}
+
+#[tokio::test]
+async fn test_request_response_notls_buffered_chunked() {
+    setup();
+    let addr = ([127, 0, 0, 1], 0).into();
+
+    let service = make_service_fn(|_| async { Ok::<_, hyper::Error>(service_fn(echo)) });
+
+    let server = Server::bind(&addr).serve(service);
+    let addr = server.local_addr();
+
+    let (tx, rx) = oneshot::channel();
+    let t = tokio::spawn(async move {
+        tokio::select! {
+            _ = server => {}
+            _ = rx => {}
+        }
+    });
+
+    let url = format!("http://127.0.0.1:{}", addr.port());
+    let mut client = HttpClient::new(&TCP, &LOOPBACK_DNS);
+    let mut tx_buf = [0; 4096];
+    let mut rx_buf = [0; 4096];
+    static CHUNKS: [&'static [u8]; 2] = [b"PART1", b"PART2"];
+    let mut request = client
+        .request(Method::POST, &url)
+        .await
+        .unwrap()
+        .into_buffered(&mut tx_buf)
+        .body(ChunkedBody(&CHUNKS))
+        .content_type(ContentType::TextPlain);
+    let response = request.send(&mut rx_buf).await.unwrap();
+    let body = response.body().read_to_end().await;
+    assert_eq!(body.unwrap(), b"PART1PART2");
+
+    tx.send(()).unwrap();
+    t.await.unwrap();
+}
+
 fn load_certs(filename: &std::path::PathBuf) -> Vec<rustls::Certificate> {
     let certfile = std::fs::File::open(filename).expect("cannot open certificate file");
     let mut reader = std::io::BufReader::new(certfile);