diff --git a/Cargo.lock b/Cargo.lock index 6d421c0f..1fca162c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1326,6 +1326,7 @@ name = "impit-node" version = "0.0.0" dependencies = [ "bytes", + "futures-util", "h2", "impit", "napi", diff --git a/impit-node/Cargo.toml b/impit-node/Cargo.toml index 26c57945..8df73f1f 100644 --- a/impit-node/Cargo.toml +++ b/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" diff --git a/impit-node/index.wrapper.js b/impit-node/index.wrapper.js index 3e9ed22e..91b74fd4 100644 --- a/impit-node/index.wrapper.js +++ b/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; } } diff --git a/impit-node/src/lib.rs b/impit-node/src/lib.rs index 90ec326d..5fa5be65 100644 --- a/impit-node/src/lib.rs +++ b/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_byte_stream())), + (bytes, None) => bytes.map(|array| array.to_vec().into()), + }); let response = if matches!(method, HttpMethod::Get | HttpMethod::Head) && body.is_some() { Err(ImpitError::BindingPassthroughError( diff --git a/impit-node/src/request.rs b/impit-node/src/request.rs index 59657050..05f3d244 100644 --- a/impit-node/src/request.rs +++ b/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); + +impl FromNapiValue for BodyStream { + unsafe fn from_napi_value(env: sys::napi_env, value: sys::napi_value) -> napi::Result { + let stream = unsafe { ReadableStream::::from_napi_value(env, value)? }; + Ok(Self(stream.read()?)) + } +} + +impl BodyStream { + pub fn into_byte_stream(self) -> impl Stream> { + 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, + /// 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, /// Request timeout in milliseconds. Overrides the Impit-wide timeout option from {@link ImpitOptions.timeout}. pub timeout: Option, /// 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. diff --git a/impit-node/test/basics.test.ts b/impit-node/test/basics.test.ts index f009f8cc..c6acbbe6 100644 --- a/impit-node/test/basics.test.ts +++ b/impit-node/test/basics.test.ts @@ -522,6 +522,49 @@ describe.each([ expect(json.data).toEqual(STRING_PAYLOAD); }); + // https://github.com/apify/impit/issues/513 + test('streamed bodies are not buffered before the request is sent', async () => { + let bodyStarted: (started: boolean) => void = () => {}; + const firstChunkArrived = new Promise((resolve) => { bodyStarted = resolve; }); + const giveUp = setTimeout(() => bodyStarted(false), 5e3); + + // Echoes back how the body was framed and what arrived, and reports the first body byte + // while the request is still in flight. + const server = http.createServer((req, res) => { + const received: Buffer[] = []; + req.on('data', (chunk) => { received.push(chunk); bodyStarted(true); }); + req.on('end', () => res.end(JSON.stringify({ + transferEncoding: req.headers['transfer-encoding'] ?? null, + contentLength: req.headers['content-length'] ?? null, + data: Buffer.concat(received).toString('utf8'), + }))); + }).listen(0); + + const sentBeforeNextChunk: boolean[] = []; + async function* chunks() { + yield '{"Impit-Test":'; + sentBeforeNextChunk.push(await firstChunkArrived); + yield '"foořžš"}'; + } + + try { + const response = await impit.fetch(`http://localhost:${(server.address() as AddressInfo).port}/`, { + method: HttpMethod.Post, + body: chunks() as any, + }); + + expect(sentBeforeNextChunk).toEqual([true]); + expect(await response.json()).toEqual({ + transferEncoding: 'chunked', + contentLength: null, + data: STRING_PAYLOAD, + }); + } finally { + clearTimeout(giveUp); + server.close(); + } + }); + test('Request with body preserves its Content-Type', async () => { const request = new Request(localPostUrl, { method: 'POST',