chore: add testing for streaming

worker-sandbox/Cargo.toml

@@ -31,6 +31,13 @@ worker = { path = "../worker", version = "0.0.9" }
 futures = "0.3.21"
 
 [dev-dependencies]
-reqwest = { version = "0.11.9", features = ["blocking", "json", "multipart"] }
+futures = "0.3.21"
+reqwest = { version = "0.11.9", features = [
+    "blocking",
+    "json",
+    "multipart",
+    "stream",
+] }
+tokio = { version = "1.16.1", features = ["macros", "rt", "test-util"] }
 tungstenite = "0.16.0"
 wasm-bindgen-test = "0.2"

worker-sandbox/tests/requests.rs

@@ -1,10 +1,12 @@
+use futures::{channel::mpsc, SinkExt, StreamExt};
 use http::StatusCode;
 use reqwest::{
     blocking::{
         multipart::{Form, Part},
         Client,
     },
     redirect::Policy,
+    Body, Client as AsyncClient,
 };
 use serde::{Deserialize, Serialize};
 use util::*;
@@ -302,3 +304,49 @@ fn custom_response_body() {
     let body = get("custom-response-body", |r| r).bytes().unwrap();
     assert_eq!(body.to_vec(), b"hello");
 }
+
+#[tokio::test]
+async fn xor() {
+    expect_wrangler();
+
+    let (mut body_sink, rx) = mpsc::channel::<u8>(32);
+    let req_stream = rx.map(|byte| Ok::<Vec<u8>, std::io::Error>(vec![byte]));
+    let body = Body::wrap_stream(req_stream);
+
+    let xor_num = 10;
+
+    // We need to send a single byte for us to get the initial response.
+    body_sink.send(0).await.unwrap();
+
+    let client = AsyncClient::new();
+    let mut res_stream = client
+        .post(&format!("http://127.0.0.1:8787/xor/{xor_num}"))
+        .body(body)
+        .send()
+        .await
+        .expect("could not make request")
+        .bytes_stream();
+
+    // Skip that first byte we use to get the stream.
+    let _ = res_stream.next().await;
+
+    for byte in 0..=255u8 {
+        body_sink.send(byte).await.unwrap();
+        let xored_byte = res_stream
+            .next()
+            .await
+            .expect("XOR stream closed unexpectedly")
+            .map(|chunk| chunk[0])
+            .expect("unexpected error with response stream");
+
+        assert_eq!(xored_byte, byte ^ xor_num);
+    }
+
+    body_sink
+        .close()
+        .await
+        .expect("unable to close body stream");
+
+    // Ensure that closing our request stream ends the body.
+    assert!(res_stream.next().await.is_none());
+}

worker-sandbox/wrangler.toml

@@ -14,7 +14,7 @@ vars = { SOME_VARIABLE = "some value" }
 bindings = [{ name = "COUNTER", class_name = "Counter" }]
 
 [build]
-# command = "worker-build --dev"
+command = "cargo install -q worker-build && worker-build --release"
 
 [build.upload]
 dir = "build/worker"