feat(node): stream request bodies

impit-node/Cargo.toml

@@ -16,6 +16,7 @@ h2 = "0.4.7"
 reqwest = { version="0.13.1" }
 tokio-stream = "0.1.17"
 bytes = "1.11.1"
+futures-util = "0.3"
 
 [build-dependencies]
 napi-build = "2.3.2"

impit-node/index.wrapper.js

@@ -49,15 +49,22 @@ function toUint8Array(chunk) {
     return typeof chunk === 'string' ? new TextEncoder().encode(chunk) : new Uint8Array(chunk);
 }
 
-function concatUint8Arrays(chunks) {
-    const totalLength = chunks.reduce((acc, chunk) => acc + chunk.length, 0);
-    const result = new Uint8Array(totalLength);
-    let offset = 0;
-    for (const chunk of chunks) {
-        result.set(chunk, offset);
-        offset += chunk.length;
-    }
-    return result;
+// Streamed bodies are passed to the native layer as a ReadableStream of bytes, so that the chunks
+// reach the connection as they are produced instead of being buffered before the request starts.
+function toByteStream(source) {
+    const reader = source instanceof ReadableStream ? source.getReader() : source[Symbol.asyncIterator]();
+    const next = reader.read?.bind(reader) ?? reader.next.bind(reader);
+
+    return new ReadableStream({
+        async pull(controller) {
+            const { done, value } = await next();
+            if (done) {
+                controller.close();
+            } else {
+                controller.enqueue(toUint8Array(value));
+            }
+        },
+    });
 }
 
 function shouldRewriteRedirectToGet(httpStatus, method) {
@@ -193,23 +200,9 @@ class Impit extends native.Impit {
             return { body: new Uint8Array(await body.arrayBuffer()), type: body.type };
         } else if (body instanceof FormData) {
             return await this.#generateMultipartFormData(body);
-        } else if (body instanceof ReadableStream) {
-            const reader = body.getReader();
-            const chunks = [];
-            let done = false;
-            while (!done) {
-                const { done: streamDone, value } = await reader.read();
-                done = streamDone;
-                if (value != null) chunks.push(toUint8Array(value));
-            }
-            return { body: concatUint8Arrays(chunks), type: '' };
-        } else if (body && typeof body[Symbol.asyncIterator] === 'function') {
-            // Node.js streams (e.g. Readable.from(...)) and other async iterables.
-            const chunks = [];
-            for await (const chunk of body) {
-                chunks.push(toUint8Array(chunk));
-            }
-            return { body: concatUint8Arrays(chunks), type: '' };
+        } else if (body instanceof ReadableStream || typeof body[Symbol.asyncIterator] === 'function') {
+            // Web streams, Node.js streams (e.g. Readable.from(...)) and other async iterables.
+            return { bodyStream: toByteStream(body), type: '' };
         }
         return { body, type: '' };
     }
@@ -238,8 +231,9 @@ class Impit extends native.Impit {
         options.headers = canonicalizeHeaders(options?.headers);
 
         if (options?.body) {
-            const { body: requestBody, type } = await this.#serializeBody(options.body);
+            const { body: requestBody, bodyStream, type } = await this.#serializeBody(options.body);
             options.body = requestBody;
+            options.bodyStream = bodyStream;
             if (type && !options.headers.some(([key]) => key.toLowerCase() === 'content-type')) {
                 options.headers.push(['Content-Type', type]);
             }
@@ -252,6 +246,7 @@ class Impit extends native.Impit {
             method: options.method,
             headers: options.headers,
             body: options.body,
+            bodyStream: options.bodyStream,
             timeout: options.timeout,
             forceHttp3: options.forceHttp3,
             signal: options.signal,
@@ -316,6 +311,7 @@ class Impit extends native.Impit {
                 method,
                 headers,
                 body: method === 'GET' ? undefined : options.body,
+                bodyStream: method === 'GET' ? undefined : options.bodyStream,
             });
 
             const originalResponse = await Promise.race([
@@ -349,6 +345,10 @@ class Impit extends native.Impit {
                     url = new URL(location, url).toString();
                     method = shouldRewriteRedirectToGet(originalResponse.status, method) ? 'GET' : method;
 
+                    if (options.bodyStream && method !== 'GET') {
+                        throw new TypeError(`Cannot follow a redirect that resends a streaming request body: ${url}`);
+                    }
+
                     continue;
                 }
             }

impit-node/src/lib.rs

@@ -1,6 +1,10 @@
 use std::time::Duration;
 
-use impit::{errors::ImpitError, impit::Impit, request::RequestOptions};
+use impit::{
+  errors::ImpitError,
+  impit::Impit,
+  request::{ImpitBody, RequestOptions},
+};
 use napi::{bindgen_prelude::ObjectFinalize, Env};
 use napi_derive::napi;
 
@@ -136,9 +140,10 @@ impl ImpitWrapper {
       .as_ref()
       .and_then(|init| init.method.to_owned())
       .unwrap_or_default();
-    let body = request_init
-      .and_then(|init| init.body)
-      .map(|array| array.to_vec().into());
+    let body = request_init.and_then(|init| match (init.body, init.body_stream) {
+      (_, Some(stream)) => Some(ImpitBody::from_stream(stream.into_bytes())),
+      (bytes, None) => bytes.map(|array| array.to_vec().into()),
+    });
 
     let response = if matches!(method, HttpMethod::Get | HttpMethod::Head) && body.is_some() {
       Err(ImpitError::BindingPassthroughError(

impit-node/src/request.rs

@@ -1,7 +1,25 @@
-use napi::bindgen_prelude::Uint8Array;
+use bytes::Bytes;
+use futures_util::{Stream, TryStreamExt};
+use napi::bindgen_prelude::{sys, FromNapiValue, ReadableStream, Reader, Uint8Array};
 
 use napi_derive::napi;
 
+/// A JS `ReadableStream` of byte chunks, read chunk by chunk as the request body is sent.
+pub struct BodyStream(Reader<Uint8Array>);
+
+impl FromNapiValue for BodyStream {
+  unsafe fn from_napi_value(env: sys::napi_env, value: sys::napi_value) -> napi::Result<Self> {
+    let stream = unsafe { ReadableStream::<Uint8Array>::from_napi_value(env, value)? };
+    Ok(Self(stream.read()?))
+  }
+}
+
+impl BodyStream {
+  pub fn into_bytes(self) -> impl Stream<Item = napi::Result<Bytes>> {
+    self.0.map_ok(|chunk| Bytes::copy_from_slice(&chunk))
+  }
+}
+
 #[derive(Default, Clone)]
 #[napi(string_enum = "UPPERCASE")]
 pub enum HttpMethod {
@@ -24,7 +42,7 @@ pub enum HttpMethod {
 ///
 /// See {@link Impit.fetch} for usage.
 #[derive(Default)]
-#[napi(object)]
+#[napi(object, object_to_js = false)]
 pub struct RequestInit {
   /// HTTP method to use for the request. Default is `GET`.
   ///
@@ -47,6 +65,9 @@ pub struct RequestInit {
   )]
   /// Request body. Can be a string, Buffer, ArrayBuffer, TypedArray, DataView, Blob, File, URLSearchParams, FormData or ReadableStream.
   pub body: Option<Uint8Array>,
+  /// Set by the JS wrapper instead of `body` when the body is a stream. Takes precedence over `body`.
+  #[napi(skip_typescript)]
+  pub body_stream: Option<BodyStream>,
   /// Request timeout in milliseconds. Overrides the Impit-wide timeout option from {@link ImpitOptions.timeout}.
   pub timeout: Option<u32>,
   /// Force the request to use HTTP/3. If the server doesn't expect HTTP/3 or the Impit instance doesn't have HTTP/3 enabled (via the {@link ImpitOptions.http3} option), the request will fail.

impit-node/test/basics.test.ts

@@ -522,6 +522,20 @@ describe.each([
             expect(json.data).toEqual(STRING_PAYLOAD);
         });
 
+        // https://github.com/apify/impit/issues/513
+        test('streamed bodies are sent chunk by chunk', async () => {
+            const chunks = ['{"Impit-Test":', '"foořžš"}'];
+            const response = await impit.fetch(localPostUrl, {
+                method: HttpMethod.Post,
+                body: Readable.from(chunks) as any,
+            });
+            const json = await response.json();
+
+            expect(json.headers?.['transfer-encoding']).toBe('chunked');
+            expect(json.headers?.['content-length']).toBeUndefined();
+            expect(json.data).toEqual(STRING_PAYLOAD);
+        });
+
         test('Request with body preserves its Content-Type', async () => {
             const request = new Request(localPostUrl, {
                 method: 'POST',