Commit d71a464
stream: add bytes() method to stream/consumers
- Add bytes() method to get Uint8Array from streams
- Add tests for bytes() method in PassThrough and ObjectMode scenarios
- Update documentation
Fixes: #59542
PR-URL: #60426
Reviewed-By: Ethan Arrowood <ethan@arrowood.dev>
Reviewed-By: René <contact.9a5d6388@renegade334.me.uk>
Reviewed-By: Jake Yuesong Li <jake.yuesong@gmail.com>
Reviewed-By: Mattias Buelens <mattias@buelens.com>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Gürgün Dayıoğlu <hey@gurgun.day>1 parent 06ce005 commit d71a464
3 files changed
Lines changed: 78 additions & 0 deletions
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -1688,6 +1688,43 @@ buffer(readable).then((data) => { | |||
| 1688 | 1688 | }); | |
| 1689 | 1689 | ``` | |
| 1690 | 1690 | ||
| 1691 | + #### `streamConsumers.bytes(stream)` | ||
| 1692 | + | ||
| 1693 | + <!-- YAML | ||
| 1694 | + added: REPLACEME | ||
| 1695 | + --> | ||
| 1696 | + | ||
| 1697 | + * `stream` {ReadableStream|stream.Readable|AsyncIterator} | ||
| 1698 | + * Returns: {Promise} Fulfills with a {Uint8Array} containing the full | ||
| 1699 | + contents of the stream. | ||
| 1700 | + | ||
| 1701 | + ```mjs | ||
| 1702 | + import { bytes } from 'node:stream/consumers'; | ||
| 1703 | + import { Readable } from 'node:stream'; | ||
| 1704 | + import { Buffer } from 'node:buffer'; | ||
| 1705 | + | ||
| 1706 | + const dataBuffer = Buffer.from('hello world from consumers!'); | ||
| 1707 | + | ||
| 1708 | + const readable = Readable.from(dataBuffer); | ||
| 1709 | + const data = await bytes(readable); | ||
| 1710 | + console.log(`from readable: ${data.length}`); | ||
| 1711 | + // Prints: from readable: 27 | ||
| 1712 | + ``` | ||
| 1713 | + | ||
| 1714 | + ```cjs | ||
| 1715 | + const { bytes } = require('node:stream/consumers'); | ||
| 1716 | + const { Readable } = require('node:stream'); | ||
| 1717 | + const { Buffer } = require('node:buffer'); | ||
| 1718 | + | ||
| 1719 | + const dataBuffer = Buffer.from('hello world from consumers!'); | ||
| 1720 | + | ||
| 1721 | + const readable = Readable.from(dataBuffer); | ||
| 1722 | + bytes(readable).then((data) => { | ||
| 1723 | + console.log(`from readable: ${data.length}`); | ||
| 1724 | + // Prints: from readable: 27 | ||
| 1725 | + }); | ||
| 1726 | + ``` | ||
| 1727 | + | ||
| 1691 | 1728 | #### `streamConsumers.json(stream)` | |
| 1692 | 1729 | ||
| 1693 | 1730 | <!-- YAML | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -2,6 +2,7 @@ | |||
| 2 | 2 | ||
| 3 | 3 | const { | |
| 4 | 4 | JSONParse, | |
| 5 | + Uint8Array, | ||
| 5 | 6 | } = primordials; | |
| 6 | 7 | ||
| 7 | 8 | const { | |
@@ -50,6 +51,14 @@ async function buffer(stream) { | |||
| 50 | 51 | return Buffer.from(await arrayBuffer(stream)); | |
| 51 | 52 | } | |
| 52 | 53 | ||
| 54 | + /** | ||
| 55 | + * @param {AsyncIterable|ReadableStream|Readable} stream | ||
| 56 | + * @returns {Promise<Uint8Array>} | ||
| 57 | + */ | ||
| 58 | + async function bytes(stream) { | ||
| 59 | + return new Uint8Array(await arrayBuffer(stream)); | ||
| 60 | + } | ||
| 61 | + | ||
| 53 | 62 | /** | |
| 54 | 63 | * @param {AsyncIterable|ReadableStream|Readable} stream | |
| 55 | 64 | * @returns {Promise<string>} | |
@@ -82,6 +91,7 @@ module.exports = { | |||
| 82 | 91 | arrayBuffer, | |
| 83 | 92 | blob, | |
| 84 | 93 | buffer, | |
| 94 | + bytes, | ||
| 85 | 95 | text, | |
| 86 | 96 | json, | |
| 87 | 97 | }; | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -8,6 +8,7 @@ const { | |||
| 8 | 8 | arrayBuffer, | |
| 9 | 9 | blob, | |
| 10 | 10 | buffer, | |
| 11 | + bytes, | ||
| 11 | 12 | text, | |
| 12 | 13 | json, | |
| 13 | 14 | } = require('stream/consumers'); | |
@@ -61,6 +62,18 @@ const kArrayBuffer = | |||
| 61 | 62 | setTimeout(() => passthrough.end('there'), 10); | |
| 62 | 63 | } | |
| 63 | 64 | ||
| 65 | + { | ||
| 66 | + const passthrough = new PassThrough(); | ||
| 67 | + | ||
| 68 | + bytes(passthrough).then(common.mustCall((uint8arr) => { | ||
| 69 | + assert(uint8arr instanceof Uint8Array); | ||
| 70 | + assert.strictEqual(uint8arr.byteLength, 10); | ||
| 71 | + assert.deepStrictEqual(Buffer.from(uint8arr), buf); | ||
| 72 | + })); | ||
| 73 | + | ||
| 74 | + passthrough.write('hello'); | ||
| 75 | + setTimeout(() => passthrough.end('there'), 10); | ||
| 76 | + } | ||
| 64 | 77 | ||
| 65 | 78 | { | |
| 66 | 79 | const passthrough = new PassThrough(); | |
@@ -219,6 +232,24 @@ const kArrayBuffer = | |||
| 219 | 232 | stream.end({}); | |
| 220 | 233 | } | |
| 221 | 234 | ||
| 235 | + { | ||
| 236 | + const stream = new PassThrough({ | ||
| 237 | + readableObjectMode: true, | ||
| 238 | + writableObjectMode: true, | ||
| 239 | + }); | ||
| 240 | + | ||
| 241 | + bytes(stream).then(common.mustCall((uint8arr) => { | ||
| 242 | + assert(uint8arr instanceof Uint8Array); | ||
| 243 | + assert.strictEqual(uint8arr.byteLength, 30); | ||
| 244 | + assert.strictEqual( | ||
| 245 | + Buffer.from(uint8arr).toString(), | ||
| 246 | + '[object Object][object Object]'); | ||
| 247 | + })); | ||
| 248 | + | ||
| 249 | + stream.write({}); | ||
| 250 | + stream.end({}); | ||
| 251 | + } | ||
| 252 | + | ||
| 222 | 253 | { | |
| 223 | 254 | const stream = new PassThrough({ | |
| 224 | 255 | readableObjectMode: true, | |
0 commit comments