stream: utility consumers for web and node.js streams

Signed-off-by: James M Snell <jasnell@gmail.com>

PR-URL: https://github.com/nodejs/node/pull/39594
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

lib/stream/consumers.js

@@ -0,0 +1,84 @@
+'use strict';
+
+const {
+  JSONParse,
+} = primordials;
+
+const {
+  TextDecoder,
+} = require('internal/encoding');
+
+const {
+  Blob,
+} = require('internal/blob');
+
+const {
+  Buffer,
+} = require('buffer');
+
+/**
+ * @typedef {import('../internal/webstreams/readablestream').ReadableStream
+ * } ReadableStream
+ * @typedef {import('../internal/streams/readable')} Readable
+ */
+
+/**
+ * @param {AsyncIterable|ReadableStream|Readable} stream
+ * @returns {Promise<Blob>}
+ */
+async function blob(stream) {
+  const chunks = [];
+  for await (const chunk of stream)
+    chunks.push(chunk);
+  return new Blob(chunks);
+}
+
+/**
+ * @param {AsyncIterable|ReadableStream|Readable} stream
+ * @returns {Promise<ArrayBuffer>}
+ */
+async function arrayBuffer(stream) {
+  const ret = await blob(stream);
+  return ret.arrayBuffer();
+}
+
+/**
+ * @param {AsyncIterable|ReadableStream|Readable} stream
+ * @returns {Promise<Buffer>}
+ */
+async function buffer(stream) {
+  return Buffer.from(await arrayBuffer(stream));
+}
+
+/**
+ * @param {AsyncIterable|ReadableStream|Readable} stream
+ * @returns {Promise<string>}
+ */
+async function text(stream) {
+  const dec = new TextDecoder();
+  let str = '';
+  for await (const chunk of stream) {
+    if (typeof chunk === 'string')
+      str += chunk;
+    else
+      str += dec.decode(chunk, { stream: true });
+  }
+  return str;
+}
+
+/**
+ * @param {AsyncIterable|ReadableStream|Readable} stream
+ * @returns {Promise<any>}
+ */
+async function json(stream) {
+  const str = await text(stream);
+  return JSONParse(str);
+}
+
+module.exports = {
+  arrayBuffer,
+  blob,
+  buffer,
+  text,
+  json,
+};