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: https://github.com/nodejs/node/issues/59542
PR-URL: https://github.com/nodejs/node/pull/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>

doc/api/webstreams.md

@@ -1694,6 +1694,43 @@ buffer(readable).then((data) => {
 });
 ```
 
+#### `streamConsumers.bytes(stream)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+* `stream` {ReadableStream|stream.Readable|AsyncIterator}
+* Returns: {Promise} Fulfills with a {Uint8Array} containing the full
+  contents of the stream.
+
+```mjs
+import { bytes } from 'node:stream/consumers';
+import { Readable } from 'node:stream';
+import { Buffer } from 'node:buffer';
+
+const dataBuffer = Buffer.from('hello world from consumers!');
+
+const readable = Readable.from(dataBuffer);
+const data = await bytes(readable);
+console.log(`from readable: ${data.length}`);
+// Prints: from readable: 27
+```
+
+```cjs
+const { bytes } = require('node:stream/consumers');
+const { Readable } = require('node:stream');
+const { Buffer } = require('node:buffer');
+
+const dataBuffer = Buffer.from('hello world from consumers!');
+
+const readable = Readable.from(dataBuffer);
+bytes(readable).then((data) => {
+  console.log(`from readable: ${data.length}`);
+  // Prints: from readable: 27
+});
+```
+
 #### `streamConsumers.json(stream)`
 
 <!-- YAML

lib/stream/consumers.js

@@ -2,6 +2,7 @@
 
 const {
   JSONParse,
+  Uint8Array,
 } = primordials;
 
 const {
@@ -50,6 +51,14 @@ async function buffer(stream) {
   return Buffer.from(await arrayBuffer(stream));
 }
 
+/**
+ * @param {AsyncIterable|ReadableStream|Readable} stream
+ * @returns {Promise<Uint8Array>}
+ */
+async function bytes(stream) {
+  return new Uint8Array(await arrayBuffer(stream));
+}
+
 /**
  * @param {AsyncIterable|ReadableStream|Readable} stream
  * @returns {Promise<string>}
@@ -82,6 +91,7 @@ module.exports = {
   arrayBuffer,
   blob,
   buffer,
+  bytes,
   text,
   json,
 };

test/parallel/test-stream-consumers.js

@@ -8,6 +8,7 @@ const {
   arrayBuffer,
   blob,
   buffer,
+  bytes,
   text,
   json,
 } = require('stream/consumers');
@@ -61,6 +62,18 @@ const kArrayBuffer =
   setTimeout(() => passthrough.end('there'), 10);
 }
 
+{
+  const passthrough = new PassThrough();
+
+  bytes(passthrough).then(common.mustCall((uint8arr) => {
+    assert(uint8arr instanceof Uint8Array);
+    assert.strictEqual(uint8arr.byteLength, 10);
+    assert.deepStrictEqual(Buffer.from(uint8arr), buf);
+  }));
+
+  passthrough.write('hello');
+  setTimeout(() => passthrough.end('there'), 10);
+}
 
 {
   const passthrough = new PassThrough();
@@ -219,6 +232,24 @@ const kArrayBuffer =
   stream.end({});
 }
 
+{
+  const stream = new PassThrough({
+    readableObjectMode: true,
+    writableObjectMode: true,
+  });
+
+  bytes(stream).then(common.mustCall((uint8arr) => {
+    assert(uint8arr instanceof Uint8Array);
+    assert.strictEqual(uint8arr.byteLength, 30);
+    assert.strictEqual(
+      Buffer.from(uint8arr).toString(),
+      '[object Object][object Object]');
+  }));
+
+  stream.write({});
+  stream.end({});
+}
+
 {
   const stream = new PassThrough({
     readableObjectMode: true,