From 99d70fb74c6c85cf1e9bf08b388b675095182abc Mon Sep 17 00:00:00 2001 From: Sebastian Beltran Date: Fri, 10 Jul 2026 12:42:32 -0500 Subject: [PATCH 1/2] feat!: split node and web streams into separate functions --- HISTORY.md | 4 +- README.md | 19 ++++- docs/examples.md | 6 +- package.json | 2 +- src/index.ts | 162 +++++++++++++++++++++++++++-------------- test/flowing.spec.ts | 6 +- test/index.bench.ts | 8 +- test/index.spec.ts | 27 ++----- test/webstream.spec.ts | 104 ++++++++++++++------------ 9 files changed, 200 insertions(+), 138 deletions(-) diff --git a/HISTORY.md b/HISTORY.md index 23a1c8c..93172ef 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -24,9 +24,9 @@ ### 🚀 Improvements -- Support reading WHATWG `ReadableStream` (web streams) - by [@bjohansebas](https://github.com/bjohansebas) in [#148](https://github.com/stream-utils/raw-body/pull/148) and [#160](https://github.com/stream-utils/raw-body/pull/160) +- Support reading WHATWG `ReadableStream` (web streams) through the new `getRawBodyWeb` export - by [@bjohansebas](https://github.com/bjohansebas) in [#148](https://github.com/stream-utils/raw-body/pull/148), [#160](https://github.com/stream-utils/raw-body/pull/160) and [#174](https://github.com/stream-utils/raw-body/pull/174) - `fetch` `Request`/`Response` bodies, `Blob.stream()`, `TransformStream` readables, and `Readable.toWeb()` bridges can be read directly. Streams already locked, read, or cancelled error with a 500 `stream.not.readable`; client aborts are mapped to the same 400 `request.aborted` error as node streams, with the original error in `cause`; string chunks are accepted without an encoding (UTF-8 `Buffer`), but error with a 500 `stream.encoding.set` when combined with one, since the stream is already decoded; non-byte chunks (e.g. `ArrayBuffer`) error with a `TypeError`. On error the reader lock is released, but the stream is not cancelled; disposing it is up to the caller. + `fetch` `Request`/`Response` bodies, `Blob.stream()`, `TransformStream` readables, and `Readable.toWeb()` bridges can be read with `getRawBodyWeb`, which takes the same options as `getRawBody`; `getRawBody` itself keeps accepting only node streams. Streams already locked, read, or cancelled error with a 500 `stream.not.readable`; client aborts are mapped to the same 400 `request.aborted` error as node streams, with the original error in `cause`; string chunks are accepted without an encoding (UTF-8 `Buffer`), but error with a 500 `stream.encoding.set` when combined with one, since the stream is already decoded; non-byte chunks (e.g. `ArrayBuffer`) error with a `TypeError`. On error the reader lock is released, but the stream is not cancelled; disposing it is up to the caller. - Add a custom `decoder` option - by [@bjohansebas](https://github.com/bjohansebas) in [#145](https://github.com/stream-utils/raw-body/pull/145) diff --git a/README.md b/README.md index 6663401..2efd079 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ npm install raw-body ## API ```js -import getRawBody from 'raw-body' +import getRawBody, { getRawBodyWeb } from 'raw-body' ``` The package is ESM-only. CommonJS consumers can load it with @@ -32,16 +32,27 @@ available in all supported Node.js versions: ```js const getRawBody = require('raw-body').default +const { getRawBodyWeb } = require('raw-body') ``` ### getRawBody(stream, [options], [callback]) **Returns a promise if no callback specified.** -The `stream` argument can be a Node.js readable stream (like an HTTP request) -or a [WHATWG `ReadableStream`](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream) +The `stream` argument must be a Node.js readable stream +(like an HTTP request). + +### getRawBodyWeb(stream, [options], [callback]) + +**Returns a promise if no callback specified.** + +The `stream` argument must be a +[WHATWG `ReadableStream`](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream) (like the body of a `fetch` `Response`). +Both functions accept the same options and behave the same, they only +differ in the kind of stream they read. + Options: - `length` - The length of the stream. @@ -82,7 +93,7 @@ getRawBody(stream, { You can also pass a string in place of options to just specify the encoding. -If an error occurs, the stream will be paused, everything unpiped, +For node streams, if an error occurs, the stream will be paused, everything unpiped, and you are responsible for correctly disposing the stream. For HTTP requests, you may need to finish consuming the stream if you want to keep the socket open for future requests. For streams diff --git a/docs/examples.md b/docs/examples.md index d3b3633..c5f5ab1 100644 --- a/docs/examples.md +++ b/docs/examples.md @@ -48,17 +48,17 @@ app.use(function * (next) { ## Simple Hono example Hono's request is a `fetch` `Request`, so its body is a WHATWG -`ReadableStream` and can be read directly: +`ReadableStream` and can be read with `getRawBodyWeb`: ```js import { Hono } from 'hono' -import getRawBody from 'raw-body' +import { getRawBodyWeb } from 'raw-body' const app = new Hono() app.post('/', async (c) => { try { - const text = await getRawBody(c.req.raw.body, { + const text = await getRawBodyWeb(c.req.raw.body, { length: c.req.header('content-length'), limit: '1mb', encoding: 'utf-8' diff --git a/package.json b/package.json index 6f95ebf..97a9754 100644 --- a/package.json +++ b/package.json @@ -36,7 +36,7 @@ "size-limit": [ { "path": "dist/index.js", - "limit": "4.61 KB", + "limit": "4.56 KB", "ignore": [ "node:async_hooks", "node:stream" diff --git a/src/index.ts b/src/index.ts index 4df544f..02cc80c 100644 --- a/src/index.ts +++ b/src/index.ts @@ -19,7 +19,12 @@ export type Encoding = string | true /** * The stream types accepted by `getRawBody`. */ -export type RawBodyStream = NodeJS.ReadableStream | Readable | ReadableStream +export type NodeRawBodyStream = NodeJS.ReadableStream | Readable + +/** + * The stream types accepted by `getRawBodyWeb`. + */ +export type WebRawBodyStream = ReadableStream /** * A streaming decoder, turning body chunks into a string. @@ -259,51 +264,20 @@ function finish (decoder: Decoder | null, buffer: string | Buffer[], length: num } /** - * Gets the entire buffer of a stream as a `Buffer`, delivered to the - * callback. Validates the stream's length against an expected length - * and maximum limit. Ideal for parsing request bodies. - */ -function getRawBody (stream: RawBodyStream, callback: Callback): void -/** - * Gets the entire buffer of a stream decoded as a string with the - * given encoding, delivered to the callback. Validates the stream's - * length against an expected length and maximum limit. Ideal for - * parsing request bodies. - */ -function getRawBody (stream: RawBodyStream, options: Readonly | Encoding, callback: Callback): void -/** - * Gets the entire buffer of a stream as a `Buffer`, delivered to the - * callback. Validates the stream's length against an expected length - * and maximum limit. Ideal for parsing request bodies. + * The reader for a stream type: `readStream` or `readWebStream`. */ -function getRawBody (stream: RawBodyStream, options: Readonly | null, callback: Callback): void -/** - * Gets the entire buffer of a stream decoded as a string with the - * given encoding. Validates the stream's length against an expected - * length and maximum limit. Ideal for parsing request bodies. - */ -function getRawBody (stream: RawBodyStream, options: Readonly | Encoding): Promise +type ReadBody = (stream: S, encoding: string | null, length: number | null, limit: number | null, createDecoder: CreateDecoder | undefined, callback: InternalCallback) => void + /** - * Gets the entire buffer of a stream as a `Buffer`. Validates the - * stream's length against an expected length and maximum limit. - * Ideal for parsing request bodies. + * Validate the options and read the stream, shared by `getRawBody` + * and `getRawBodyWeb`. The reader is passed in, without allocating + * an intermediate closure per call. */ -function getRawBody (stream: RawBodyStream, options?: Readonly | null): Promise -function getRawBody (stream: RawBodyStream, options?: Readonly | Encoding | Callback | null, callback?: Callback | Callback): Promise | void { + +function getBody (read: ReadBody, stream: S, options: Readonly | Encoding | Callback | null | undefined, callback: Callback | Callback | undefined): Promise | void { let done = callback as InternalCallback | undefined let opts: Readonly = (options || {}) as Options - // light validation. - if (stream === undefined) { - throw new TypeError('argument stream is required') - } - - const nodeReadable = typeof stream === 'object' && stream !== null && isNodeReadable(stream) - - if (!nodeReadable && (typeof stream !== 'object' || stream === null || !isWebReadable(stream))) { - throw new TypeError('argument stream must be a stream') - } - if (options === true || typeof options === 'string') { // short cut for encoding opts = { @@ -346,36 +320,112 @@ function getRawBody (stream: RawBodyStream, options?: Readonly | Encodi : NaN const length = Number.isNaN(parsedLength) ? null : parsedLength - // dispatch to the reader for the stream type, without allocating an - // intermediate closure per call. node streams take precedence. const decoder = opts.decoder if (done) { // classic callback style - const bound = bindAsyncContext(done) - if (nodeReadable) { - readStream(stream, encoding, length, limit, decoder, bound) - } else { - readWebStream(stream, encoding, length, limit, decoder, bound) - } + read(stream, encoding, length, limit, decoder, bindAsyncContext(done)) return } return new Promise(function executor (resolve, reject) { - const onRead: InternalCallback = function onRead (err, buf) { + read(stream, encoding, length, limit, decoder, function onRead (err, buf) { if (err) return reject(err) resolve(buf as Buffer | string) - } - if (nodeReadable) { - readStream(stream, encoding, length, limit, decoder, onRead) - } else { - readWebStream(stream, encoding, length, limit, decoder, onRead) - } + }) }) } +/** + * Gets the entire buffer of a node stream as a `Buffer`, delivered to + * the callback. Validates the stream's length against an expected + * length and maximum limit. Ideal for parsing request bodies. + */ +function getRawBody (stream: NodeRawBodyStream, callback: Callback): void +/** + * Gets the entire buffer of a node stream decoded as a string with the + * given encoding, delivered to the callback. Validates the stream's + * length against an expected length and maximum limit. Ideal for + * parsing request bodies. + */ +function getRawBody (stream: NodeRawBodyStream, options: Readonly | Encoding, callback: Callback): void +/** + * Gets the entire buffer of a node stream as a `Buffer`, delivered to + * the callback. Validates the stream's length against an expected + * length and maximum limit. Ideal for parsing request bodies. + */ +function getRawBody (stream: NodeRawBodyStream, options: Readonly | null, callback: Callback): void +/** + * Gets the entire buffer of a node stream decoded as a string with the + * given encoding. Validates the stream's length against an expected + * length and maximum limit. Ideal for parsing request bodies. + */ +function getRawBody (stream: NodeRawBodyStream, options: Readonly | Encoding): Promise +/** + * Gets the entire buffer of a node stream as a `Buffer`. Validates the + * stream's length against an expected length and maximum limit. + * Ideal for parsing request bodies. + */ +function getRawBody (stream: NodeRawBodyStream, options?: Readonly | null): Promise +function getRawBody (stream: NodeRawBodyStream, options?: Readonly | Encoding | Callback | null, callback?: Callback | Callback): Promise | void { + // light validation. + if (stream === undefined) { + throw new TypeError('argument stream is required') + } + + if (typeof stream !== 'object' || stream === null || !isNodeReadable(stream)) { + throw new TypeError('argument stream must be a node stream') + } + + return getBody(readStream, stream, options, callback) +} + +/** + * Gets the entire buffer of a web `ReadableStream` as a `Buffer`, + * delivered to the callback. Validates the stream's length against an + * expected length and maximum limit. Ideal for parsing request bodies. + */ +function getRawBodyWeb (stream: WebRawBodyStream, callback: Callback): void +/** + * Gets the entire buffer of a web `ReadableStream` decoded as a string + * with the given encoding, delivered to the callback. Validates the + * stream's length against an expected length and maximum limit. Ideal + * for parsing request bodies. + */ +function getRawBodyWeb (stream: WebRawBodyStream, options: Readonly | Encoding, callback: Callback): void +/** + * Gets the entire buffer of a web `ReadableStream` as a `Buffer`, + * delivered to the callback. Validates the stream's length against an + * expected length and maximum limit. Ideal for parsing request bodies. + */ +function getRawBodyWeb (stream: WebRawBodyStream, options: Readonly | null, callback: Callback): void +/** + * Gets the entire buffer of a web `ReadableStream` decoded as a string + * with the given encoding. Validates the stream's length against an + * expected length and maximum limit. Ideal for parsing request bodies. + */ +function getRawBodyWeb (stream: WebRawBodyStream, options: Readonly | Encoding): Promise +/** + * Gets the entire buffer of a web `ReadableStream` as a `Buffer`. + * Validates the stream's length against an expected length and + * maximum limit. Ideal for parsing request bodies. + */ +function getRawBodyWeb (stream: WebRawBodyStream, options?: Readonly | null): Promise +function getRawBodyWeb (stream: WebRawBodyStream, options?: Readonly | Encoding | Callback | null, callback?: Callback | Callback): Promise | void { + // light validation. + if (stream === undefined) { + throw new TypeError('argument stream is required') + } + + if (typeof stream !== 'object' || stream === null || !isWebReadable(stream)) { + throw new TypeError('argument stream must be a web ReadableStream') + } + + return getBody(readWebStream, stream, options, callback) +} + export default getRawBody -export { getRawBody } +export { getRawBody, getRawBodyWeb } /** * Halt a stream. diff --git a/test/flowing.spec.ts b/test/flowing.spec.ts index a70aebe..dd16cbc 100644 --- a/test/flowing.spec.ts +++ b/test/flowing.spec.ts @@ -1,7 +1,7 @@ import assert from 'node:assert' import { Readable, Writable } from 'node:stream' import { describe, it } from 'vitest' -import getRawBody from '../src/index.ts' +import getRawBody, { getRawBodyWeb } from '../src/index.ts' import { withDone } from './support/with-done.ts' const defaultLimit = 1024 * 1024 @@ -114,7 +114,7 @@ describe('stream flowing', function () { it('should stop pulling', withDone(function (done) { const stream = createInfiniteWebStream() - getRawBody(stream, { + getRawBodyWeb(stream, { limit: defaultLimit }, function (err) { assert.ok(err) @@ -129,7 +129,7 @@ describe('stream flowing', function () { it('should not pull when length exceeds limit', withDone(function (done) { const stream = createInfiniteWebStream() - getRawBody(stream, { + getRawBodyWeb(stream, { length: defaultLimit * 2, limit: defaultLimit }, function (err) { diff --git a/test/index.bench.ts b/test/index.bench.ts index 58266ed..3637ba2 100644 --- a/test/index.bench.ts +++ b/test/index.bench.ts @@ -1,6 +1,6 @@ import { Readable } from 'node:stream' import { bench, describe } from 'vitest' -import getRawBody from '../src/index.ts' +import getRawBody, { getRawBodyWeb } from '../src/index.ts' /** * Compares the node stream path against the web stream path with @@ -73,7 +73,7 @@ for (const scenario of scenarios) { }) bench('web stream', async () => { - await getRawBody(webStream(chunks), options) + await getRawBodyWeb(webStream(chunks), options) }) }) @@ -83,7 +83,7 @@ for (const scenario of scenarios) { }) bench('web stream', async () => { - await getRawBody(webStream(chunks), stringOptions) + await getRawBodyWeb(webStream(chunks), stringOptions) }) }) @@ -97,7 +97,7 @@ for (const scenario of scenarios) { }) bench('web stream', async () => { - await getRawBody(webStream(chunks), noLengthOptions) + await getRawBodyWeb(webStream(chunks), noLengthOptions) }) }) } diff --git a/test/index.spec.ts b/test/index.spec.ts index fae7cac..3f9fe95 100644 --- a/test/index.spec.ts +++ b/test/index.spec.ts @@ -18,13 +18,16 @@ describe('Raw Body', function () { // @ts-expect-error missing stream argument assert.throws(function () { getRawBody() }, /argument stream is required/) // @ts-expect-error invalid stream argument - assert.throws(function () { getRawBody(null) }, /argument stream must be a stream/) + assert.throws(function () { getRawBody(null) }, /argument stream must be a node stream/) // @ts-expect-error invalid stream argument - assert.throws(function () { getRawBody(42) }, /argument stream must be a stream/) + assert.throws(function () { getRawBody(42) }, /argument stream must be a node stream/) // @ts-expect-error invalid stream argument - assert.throws(function () { getRawBody('str') }, /argument stream must be a stream/) + assert.throws(function () { getRawBody('str') }, /argument stream must be a node stream/) // @ts-expect-error invalid stream argument - assert.throws(function () { getRawBody({}) }, /argument stream must be a stream/) + assert.throws(function () { getRawBody({}) }, /argument stream must be a node stream/) + // a web ReadableStream belongs to getRawBodyWeb + // @ts-expect-error invalid stream argument + assert.throws(function () { getRawBody(new ReadableStream()) }, /argument stream must be a node stream/) }) it('should work without any options', withDone(function (done) { @@ -332,22 +335,6 @@ describe('Raw Body', function () { }) })) - it('should prefer the node stream interface when both are present', withDone(function (done) { - const stream = createStream(Buffer.from('hello, world!')) as Readable & { getReader?: () => never } - - // a wrapper library may decorate a node stream with a - // web-stream compatibility shim; the node path must win - stream.getReader = function () { - throw new Error('getReader should not be called') - } - - getRawBody(stream, { encoding: true }, function (err, str) { - assert.ifError(err) - assert.strictEqual(str, 'hello, world!') - done() - }) - })) - it('should not invoke the callback synchronously on early errors', withDone(function (done) { let returned = false diff --git a/test/webstream.spec.ts b/test/webstream.spec.ts index 7f76957..fded01f 100644 --- a/test/webstream.spec.ts +++ b/test/webstream.spec.ts @@ -5,38 +5,52 @@ import { Readable } from 'node:stream' import type { ReadableWritablePair } from 'node:stream/web' import iconv from 'iconv-lite' import { describe, it } from 'vitest' -import getRawBody, { type RawBodyError } from '../src/index.ts' +import { getRawBodyWeb, type RawBodyError } from '../src/index.ts' import { isBun } from './support/runtime.ts' import { withDone } from './support/with-done.ts' describe('using web streams', function () { + it('should validate stream', function () { + // @ts-expect-error missing stream argument + assert.throws(function () { getRawBodyWeb() }, /argument stream is required/) + // @ts-expect-error invalid stream argument + assert.throws(function () { getRawBodyWeb(null) }, /argument stream must be a web ReadableStream/) + // @ts-expect-error invalid stream argument + assert.throws(function () { getRawBodyWeb(42) }, /argument stream must be a web ReadableStream/) + // @ts-expect-error invalid stream argument + assert.throws(function () { getRawBodyWeb({}) }, /argument stream must be a web ReadableStream/) + // a node stream belongs to getRawBody + // @ts-expect-error invalid stream argument + assert.throws(function () { getRawBodyWeb(new Readable()) }, /argument stream must be a web ReadableStream/) + }) + it('should read a ReadableStream into a buffer', async function () { - const buf = await getRawBody(createWebStream(['hello', ', ', 'world!'])) + const buf = await getRawBodyWeb(createWebStream(['hello', ', ', 'world!'])) assert.ok(Buffer.isBuffer(buf)) assert.strictEqual(buf.toString(), 'hello, world!') }) it('should read an empty ReadableStream', async function () { - const buf = await getRawBody(createWebStream([])) + const buf = await getRawBodyWeb(createWebStream([])) assert.ok(Buffer.isBuffer(buf)) assert.strictEqual(buf.length, 0) }) it('should work with encoding', async function () { - const str = await getRawBody(createWebStream(['hello, world!']), { + const str = await getRawBodyWeb(createWebStream(['hello, world!']), { encoding: 'utf-8' }) assert.strictEqual(str, 'hello, world!') }) it('should work with `true` as an option', async function () { - const str = await getRawBody(createWebStream(['hello, world!']), true) + const str = await getRawBodyWeb(createWebStream(['hello, world!']), true) assert.strictEqual(str, 'hello, world!') }) it('should decode multi-byte characters split across chunks', async function () { const bytes = Buffer.from('é€好', 'utf-8') - const str = await getRawBody(createWebStream([ + const str = await getRawBodyWeb(createWebStream([ bytes.subarray(0, 3), bytes.subarray(3) ]), { @@ -46,13 +60,13 @@ describe('using web streams', function () { }) it('should work with string chunks', async function () { - const buf = await getRawBody(createWebStream(['hello', ', world!'], { binary: false })) + const buf = await getRawBodyWeb(createWebStream(['hello', ', world!'], { binary: false })) assert.ok(Buffer.isBuffer(buf)) assert.strictEqual(buf.toString(), 'hello, world!') }) it('should read a TextDecoderStream as a utf-8 Buffer', async function () { - // TextDecoderStream emits already-decoded string chunks; getRawBody + // TextDecoderStream emits already-decoded string chunks; getRawBodyWeb // re-encodes them to utf-8. multi-byte characters split across the // underlying byte chunks must survive the round trip. const bytes = Buffer.from('café ☕ 好', 'utf-8') @@ -67,7 +81,7 @@ describe('using web streams', function () { }) const stream = byteStream.pipeThrough(new TextDecoderStream()) - const buf = await getRawBody(stream) + const buf = await getRawBodyWeb(stream) assert.ok(Buffer.isBuffer(buf)) assert.strictEqual(buf.toString('utf-8'), 'café ☕ 好') @@ -78,7 +92,7 @@ describe('using web streams', function () { // re-decoding it with the declared encoding would corrupt the data const stream = createWebStream(['hello, world!'], { binary: false }) - await assert.rejects(getRawBody(stream, { encoding: 'utf-16le' }), function (err: RawBodyError) { + await assert.rejects(getRawBodyWeb(stream, { encoding: 'utf-16le' }), function (err: RawBodyError) { assert.strictEqual(err.status, 500) assert.strictEqual(err.type, 'stream.encoding.set') return true @@ -88,7 +102,7 @@ describe('using web streams', function () { }) it('should work with a custom decoder', async function () { - const str = await getRawBody(createWebStream([Buffer.from('636f6f6c20f09f9880', 'hex')]), { + const str = await getRawBodyWeb(createWebStream([Buffer.from('636f6f6c20f09f9880', 'hex')]), { encoding: 'utf-8', decoder: iconv.getDecoder.bind(iconv) }) @@ -100,7 +114,7 @@ describe('using web streams', function () { // split mid-character: each utf-32 code unit is 4 bytes, // so the decoder must carry state across chunks - const str = await getRawBody(createWebStream([ + const str = await getRawBodyWeb(createWebStream([ bytes.subarray(0, 6), bytes.subarray(6) ]), { @@ -112,7 +126,7 @@ describe('using web streams', function () { }) it('should work with the callback style', withDone(function (done) { - getRawBody(createWebStream(['hello, world!']), function (err, buf) { + getRawBodyWeb(createWebStream(['hello, world!']), function (err, buf) { assert.ifError(err) assert.strictEqual(buf.toString(), 'hello, world!') done() @@ -120,14 +134,14 @@ describe('using web streams', function () { })) it('should check length', async function () { - const buf = await getRawBody(createWebStream(['hello, world!']), { + const buf = await getRawBodyWeb(createWebStream(['hello, world!']), { length: 13 }) assert.strictEqual(buf.toString(), 'hello, world!') }) it('should error with length mismatch', async function () { - await assert.rejects(getRawBody(createWebStream(['hello, world!']), { + await assert.rejects(getRawBodyWeb(createWebStream(['hello, world!']), { length: 10 }), function (err: RawBodyError) { assert.strictEqual(err.status, 400) @@ -139,7 +153,7 @@ describe('using web streams', function () { }) it('should error when limit is exceeded', async function () { - await assert.rejects(getRawBody(createWebStream(['hello, world!']), { + await assert.rejects(getRawBodyWeb(createWebStream(['hello, world!']), { limit: 5 }), function (err: RawBodyError) { assert.strictEqual(err.status, 413) @@ -152,7 +166,7 @@ describe('using web streams', function () { it('should error early when length > limit', async function () { const stream = createWebStream(['hello, world!']) - await assert.rejects(getRawBody(stream, { + await assert.rejects(getRawBodyWeb(stream, { length: 13, limit: 5 }), function (err: RawBodyError) { @@ -166,7 +180,7 @@ describe('using web streams', function () { }) it('should error for an unsupported encoding', async function () { - await assert.rejects(getRawBody(createWebStream(['hello, world!']), { + await assert.rejects(getRawBodyWeb(createWebStream(['hello, world!']), { encoding: 'foo/bar' }), function (err: RawBodyError) { assert.strictEqual(err.status, 415) @@ -179,7 +193,7 @@ describe('using web streams', function () { const stream = createWebStream(['hello, world!']) const reader = stream.getReader() - await assert.rejects(getRawBody(stream), function (err: RawBodyError) { + await assert.rejects(getRawBodyWeb(stream), function (err: RawBodyError) { assert.strictEqual(err.status, 500) assert.strictEqual(err.type, 'stream.not.readable') return true @@ -192,7 +206,7 @@ describe('using web streams', function () { const res = new Response('hello, world!') await res.text() - await assert.rejects(getRawBody(res.body!), function (err: RawBodyError) { + await assert.rejects(getRawBodyWeb(res.body!), function (err: RawBodyError) { assert.strictEqual(err.status, 500) assert.strictEqual(err.type, 'stream.not.readable') return true @@ -203,7 +217,7 @@ describe('using web streams', function () { const stream = createWebStream(['hello, world!']) const piped = stream.pipeThrough(new TextDecoderStream() as ReadableWritablePair) - await assert.rejects(getRawBody(stream), function (err: RawBodyError) { + await assert.rejects(getRawBodyWeb(stream), function (err: RawBodyError) { assert.strictEqual(err.status, 500) assert.strictEqual(err.type, 'stream.not.readable') return true @@ -218,7 +232,7 @@ describe('using web streams', function () { const stream = createWebStream(['hello, world!']) await stream.cancel() - const buf = await getRawBody(stream) + const buf = await getRawBodyWeb(stream) assert.ok(Buffer.isBuffer(buf)) assert.strictEqual(buf.length, 0) }) @@ -229,7 +243,7 @@ describe('using web streams', function () { while (!(await reader.read()).done); reader.releaseLock() - const buf = await getRawBody(stream) + const buf = await getRawBodyWeb(stream) assert.ok(Buffer.isBuffer(buf)) assert.strictEqual(buf.length, 0) }) @@ -248,7 +262,7 @@ describe('using web streams', function () { } }) - await assert.rejects(getRawBody(stream, { length: 10 }), function (err: RawBodyError) { + await assert.rejects(getRawBodyWeb(stream, { length: 10 }), function (err: RawBodyError) { assert.strictEqual(err.status, 400) assert.strictEqual(err.type, 'request.aborted') assert.strictEqual(err.code, 'ECONNABORTED') @@ -263,7 +277,7 @@ describe('using web streams', function () { let clientRequest: http.ClientRequest const server = http.createServer(function (req, res) { - getRawBody(Readable.toWeb(req), { + getRawBodyWeb(Readable.toWeb(req), { length: req.headers['content-length'], limit: '1kb' }, function (err) { @@ -304,7 +318,7 @@ describe('using web streams', function () { const socket = net.connect((server.address() as net.AddressInfo).port) socket.on('connect', function () { - getRawBody(Readable.toWeb(socket), function (err) { + getRawBodyWeb(Readable.toWeb(socket), function (err) { server.close() assert.ok(err) @@ -321,7 +335,7 @@ describe('using web streams', function () { const nodeStream = new Readable({ read () {} }) const webStream = Readable.toWeb(nodeStream) - getRawBody(webStream, function (err) { + getRawBodyWeb(webStream, function (err) { assert.ok(err) assert.strictEqual(err.status, 400) assert.strictEqual(err.type, 'request.aborted') @@ -341,13 +355,13 @@ describe('using web streams', function () { } }) - await assert.rejects(getRawBody(stream), /boom/) + await assert.rejects(getRawBodyWeb(stream), /boom/) }) it('should error when the decoder throws while writing', async function () { const stream = createWebStream(['hello, world!']) - await assert.rejects(getRawBody(stream, { + await assert.rejects(getRawBodyWeb(stream, { encoding: 'utf-8', decoder: function () { return { @@ -365,7 +379,7 @@ describe('using web streams', function () { it('should error when the decoder throws at the end', async function () { const stream = createWebStream(['hello, world!']) - await assert.rejects(getRawBody(stream, { + await assert.rejects(getRawBodyWeb(stream, { encoding: 'utf-8', decoder: function () { return { @@ -387,7 +401,7 @@ describe('using web streams', function () { } }) - await assert.rejects(getRawBody(stream), function (err: Error) { + await assert.rejects(getRawBodyWeb(stream), function (err: Error) { assert.ok(err instanceof Error) assert.strictEqual(err.message, 'stream error') assert.strictEqual(err.cause, 'timeout') @@ -402,7 +416,7 @@ describe('using web streams', function () { } }) - await assert.rejects(getRawBody(stream), /stream error/) + await assert.rejects(getRawBodyWeb(stream), /stream error/) }) it('should error when the stream yields an invalid chunk', async function () { @@ -413,7 +427,7 @@ describe('using web streams', function () { } }) - await assert.rejects(getRawBody(stream), TypeError) + await assert.rejects(getRawBodyWeb(stream), TypeError) // the lock is released, so the rest of the stream can be handled assert.strictEqual(stream.locked, false) @@ -429,7 +443,7 @@ describe('using web streams', function () { } }) - await assert.rejects(getRawBody(stream, { limit: 5 }), /Uint8Array or string/) + await assert.rejects(getRawBodyWeb(stream, { limit: 5 }), /Uint8Array or string/) assert.strictEqual(stream.locked, false) }) @@ -437,20 +451,20 @@ describe('using web streams', function () { const chunk = Buffer.alloc(768 * 1024, 0x61) const big = Buffer.concat([chunk, chunk]) - const buf = await getRawBody(createWebStream([chunk, chunk]), { length: big.length }) + const buf = await getRawBodyWeb(createWebStream([chunk, chunk]), { length: big.length }) assert.strictEqual(buf.length, big.length) assert.ok(buf.equals(big)) }) it('should release the lock when finished', async function () { const stream = createWebStream(['hello, world!']) - await getRawBody(stream) + await getRawBodyWeb(stream) assert.strictEqual(stream.locked, false) }) it('should read the body of a fetch Response', async function () { const res = new Response('hello, world!') - const str = await getRawBody(res.body!, { + const str = await getRawBodyWeb(res.body!, { encoding: 'utf-8', limit: '1kb' }) @@ -459,7 +473,7 @@ describe('using web streams', function () { it('should read the stream of a Blob', async function () { const blob = new Blob(['hello, world!']) - const str = await getRawBody(blob.stream(), { + const str = await getRawBodyWeb(blob.stream(), { encoding: 'utf-8' }) assert.strictEqual(str, 'hello, world!') @@ -470,7 +484,7 @@ describe('using web streams', function () { method: 'POST', body: 'hello, world!' }) - const str = await getRawBody(req.body!, { + const str = await getRawBodyWeb(req.body!, { encoding: 'utf-8' }) assert.strictEqual(str, 'hello, world!') @@ -495,7 +509,7 @@ describe('using web streams', function () { setTimeout(function () { controller.abort() }, 20) - await assert.rejects(getRawBody(res.body!, { length: 100 }), function (err: RawBodyError) { + await assert.rejects(getRawBodyWeb(res.body!, { length: 100 }), function (err: RawBodyError) { assert.strictEqual(err.status, 400) assert.strictEqual(err.type, 'request.aborted') assert.strictEqual(err.code, 'ECONNABORTED') @@ -510,10 +524,10 @@ describe('using web streams', function () { }) it('should read the readable side of a TransformStream', async function () { - const gzipped = await getRawBody( + const gzipped = await getRawBodyWeb( new Blob(['hello, world!']).stream().pipeThrough(new CompressionStream('gzip')) ) - const str = await getRawBody( + const str = await getRawBodyWeb( new Blob([new Uint8Array(gzipped)]).stream().pipeThrough(new DecompressionStream('gzip')), { encoding: 'utf-8' } ) @@ -523,7 +537,7 @@ describe('using web streams', function () { it('should not invoke the callback synchronously on early errors', withDone(function (done) { let returned = false - getRawBody(createWebStream(['hello, world!']), { + getRawBodyWeb(createWebStream(['hello, world!']), { length: 13, limit: 5 }, function (err) { @@ -553,7 +567,7 @@ describe('using web streams', function () { timer = setTimeout(reject, 1000, new Error('expected an unhandled rejection')) }) - getRawBody(createWebStream(['hello, world!']), function () { + getRawBodyWeb(createWebStream(['hello, world!']), function () { calls++ throw failure }) @@ -577,7 +591,7 @@ describe('using web streams', function () { it('should release the lock on error', async function () { const stream = createWebStream(['hello', ', world!']) - await assert.rejects(getRawBody(stream, { limit: 3 })) + await assert.rejects(getRawBodyWeb(stream, { limit: 3 })) // the lock is released, so the rest of the stream can be handled assert.strictEqual(stream.locked, false) From c126c1f95fb73444e7c743b95290afa05ec747ba Mon Sep 17 00:00:00 2001 From: Sebastian Beltran Date: Fri, 10 Jul 2026 12:48:48 -0500 Subject: [PATCH 2/2] Apply suggestion from @bjohansebas --- HISTORY.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/HISTORY.md b/HISTORY.md index 93172ef..b2cbdae 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -24,7 +24,7 @@ ### 🚀 Improvements -- Support reading WHATWG `ReadableStream` (web streams) through the new `getRawBodyWeb` export - by [@bjohansebas](https://github.com/bjohansebas) in [#148](https://github.com/stream-utils/raw-body/pull/148), [#160](https://github.com/stream-utils/raw-body/pull/160) and [#174](https://github.com/stream-utils/raw-body/pull/174) +- Support reading WHATWG `ReadableStream` (web streams) through the new `getRawBodyWeb` export - by [@bjohansebas](https://github.com/bjohansebas) in [#148](https://github.com/stream-utils/raw-body/pull/148), [#160](https://github.com/stream-utils/raw-body/pull/160) and [#175](https://github.com/stream-utils/raw-body/pull/175) `fetch` `Request`/`Response` bodies, `Blob.stream()`, `TransformStream` readables, and `Readable.toWeb()` bridges can be read with `getRawBodyWeb`, which takes the same options as `getRawBody`; `getRawBody` itself keeps accepting only node streams. Streams already locked, read, or cancelled error with a 500 `stream.not.readable`; client aborts are mapped to the same 400 `request.aborted` error as node streams, with the original error in `cause`; string chunks are accepted without an encoding (UTF-8 `Buffer`), but error with a 500 `stream.encoding.set` when combined with one, since the stream is already decoded; non-byte chunks (e.g. `ArrayBuffer`) error with a `TypeError`. On error the reader lock is released, but the stream is not cancelled; disposing it is up to the caller.