From d44775c68ff456d91ffb27903cffc456ced8e730 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Sun, 25 Jul 2021 21:00:33 +0200
Subject: [PATCH 01/17] stream: implement fetch body mixin on Readable

Make Readable exposew the fetch boxy mixin API.

Bypasses webstream glue when possible.

Refs: https://fetch.spec.whatwg.org/#body-mixin
---
 .../streams/readable-readstring-async.js      |  37 ++
 benchmark/streams/readable-readstring-data.js |  32 ++
 .../streams/readable-readstring-native.js     |  25 ++
 benchmark/streams/readable-readstring-text.js |  28 ++
 doc/api/stream.md                             |  61 +++
 lib/internal/streams/readable.js              | 340 ++++++++++++++++-
 .../test-stream-readable-body-mixin.js        | 355 ++++++++++++++++++
 7 files changed, 859 insertions(+), 19 deletions(-)
 create mode 100644 benchmark/streams/readable-readstring-async.js
 create mode 100644 benchmark/streams/readable-readstring-data.js
 create mode 100644 benchmark/streams/readable-readstring-native.js
 create mode 100644 benchmark/streams/readable-readstring-text.js
 create mode 100644 test/parallel/test-stream-readable-body-mixin.js

diff --git a/benchmark/streams/readable-readstring-async.js b/benchmark/streams/readable-readstring-async.js
new file mode 100644
index 00000000000000..f2eff78ba575f2
--- /dev/null
+++ b/benchmark/streams/readable-readstring-async.js
@@ -0,0 +1,37 @@
+'use strict';
+
+const common = require('../common');
+const Readable = require('stream').Readable;
+
+const bench = common.createBenchmark(main, {
+  n: [50e2]
+});
+
+function main({ n }) {
+  const b = 'some random string';
+  const s = new Readable();
+  function noop() {}
+  s._read = noop;
+
+  bench.start();
+
+  text(s).then((ret) => {
+    console.log(ret.length);
+    bench.end(n, ret);
+  });
+
+  for (let k = 0; k < n; ++k) {
+    for (let i = 0; i < 1e3; ++i)
+      s.push(b);
+  }
+  s.push(null);
+}
+
+async function text(s) {
+  s.setEncoding('utf8');
+  let ret = '';
+  for await (const chunk of s) {
+    ret += chunk;
+  }
+  return ret;
+}
diff --git a/benchmark/streams/readable-readstring-data.js b/benchmark/streams/readable-readstring-data.js
new file mode 100644
index 00000000000000..03d9e36d38befc
--- /dev/null
+++ b/benchmark/streams/readable-readstring-data.js
@@ -0,0 +1,32 @@
+'use strict';
+
+const common = require('../common');
+const Readable = require('stream').Readable;
+
+const bench = common.createBenchmark(main, {
+  n: [50e2]
+});
+
+function main({ n }) {
+  const b = 'some random string';
+  const s = new Readable();
+  function noop() {}
+  s._read = noop;
+
+  bench.start();
+
+  let ret = '';
+  s.setEncoding('utf8');
+  s.on('data', (chunk) => {
+    ret += chunk;
+  }).on('end', () => {
+    console.log(ret.length);
+    bench.end(n, ret);
+  });
+
+  for (let k = 0; k < n; ++k) {
+    for (let i = 0; i < 1e3; ++i)
+      s.push(b);
+  }
+  s.push(null);
+}
diff --git a/benchmark/streams/readable-readstring-native.js b/benchmark/streams/readable-readstring-native.js
new file mode 100644
index 00000000000000..1374bcf53aff76
--- /dev/null
+++ b/benchmark/streams/readable-readstring-native.js
@@ -0,0 +1,25 @@
+'use strict';
+
+const common = require('../common');
+const Readable = require('stream').Readable;
+
+const bench = common.createBenchmark(main, {
+  n: [50e2]
+});
+
+function main({ n }) {
+  const b = 'some random string';
+  const s = new Readable();
+  function noop() {}
+  s._read = noop;
+
+  bench.start();
+
+  let ret = '';
+  for (let k = 0; k < n; ++k) {
+    for (let i = 0; i < 1e3; ++i)
+      ret += b;
+  }
+  console.log(ret.length);
+  bench.end(n);
+}
diff --git a/benchmark/streams/readable-readstring-text.js b/benchmark/streams/readable-readstring-text.js
new file mode 100644
index 00000000000000..eba63f4077646e
--- /dev/null
+++ b/benchmark/streams/readable-readstring-text.js
@@ -0,0 +1,28 @@
+'use strict';
+
+const common = require('../common');
+const Readable = require('stream').Readable;
+
+const bench = common.createBenchmark(main, {
+  n: [50e2]
+});
+
+function main({ n }) {
+  const b = 'some random string';
+  const s = new Readable();
+  function noop() {}
+  s._read = noop;
+
+  bench.start();
+
+  s.text().then((ret) => {
+    console.log(ret.length);
+    bench.end(n);
+  });
+
+  for (let k = 0; k < n; ++k) {
+    for (let i = 0; i < 1e3; ++i)
+      s.push(b);
+  }
+  s.push(null);
+}
diff --git a/doc/api/stream.md b/doc/api/stream.md
index d616bb78bccfd0..e4a3db97b1b0d8 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -814,6 +814,8 @@ added: v0.9.4
 
 <!--type=class-->
 
+`Readable` implements the body mixin API from the [fetch specification][].
+
 ##### Event: `'close'`
 <!-- YAML
 added: v0.9.4
@@ -986,6 +988,44 @@ added: v0.9.4
 The `'resume'` event is emitted when [`stream.resume()`][stream-resume] is
 called and `readableFlowing` is not `true`.
 
+##### `readable.arrayBuffer()`
+<!--- YAML>
+added: REPLACEME
+--->
+
+* Returns: {Promise}
+
+Returns a promise resolving to the full content as an array buffer.
+
+##### `readable.blob()`
+<!--- YAML>
+added: REPLACEME
+--->
+
+* Returns: {Promise}
+
+Returns a promise resolving to the full content read as
+Blob.
+
+##### `readable.body`
+<!--- YAML>
+added: REPLACEME
+--->
+
+* Returns: {ReadableStream}
+
+Returns a `ReadableStream`.
+
+##### `readable.bodyUsed`
+<!--- YAML>
+added: REPLACEME
+--->
+
+* Returns: {Boolean}
+
+Returns true if stream has emitted `'data'`, `'end'`, `'error'` or
+`'close'`.
+
 ##### `readable.destroy([error])`
 <!-- YAML
 added: v8.0.0
@@ -1040,6 +1080,16 @@ readable.resume();
 readable.isPaused(); // === false
 ```
 
+##### `readable.json()`
+<!--- YAML>
+added: REPLACEME
+--->
+
+* Returns: {Promise}
+
+Returns a promise resolving to the full content read as
+utf8 string and parsed to JSON object.
+
 ##### `readable.pause()`
 <!-- YAML
 added: v0.9.4
@@ -1341,6 +1391,16 @@ readable.on('data', (chunk) => {
 });
 ```
 
+##### `readable.text()`
+<!--- YAML>
+added: REPLACEME
+--->
+
+* Returns: {Promise}
+
+Returns a promise resolving to the full content read as
+utf8 string.
+
 ##### `readable.unpipe([destination])`
 <!-- YAML
 added: v0.9.4
@@ -3503,6 +3563,7 @@ contain multi-byte characters.
 [child process stdin]: child_process.md#child_process_subprocess_stdin
 [child process stdout and stderr]: child_process.md#child_process_subprocess_stdout
 [crypto]: crypto.md
+[fetch specification]: https://fetch.spec.whatwg.org/#body-mixin
 [fs read streams]: fs.md#fs_class_fs_readstream
 [fs write streams]: fs.md#fs_class_fs_writestream
 [http-incoming-message]: http.md#http_class_http_incomingmessage
diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index e7dd60a6d78c13..f8060708730a2c 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -22,17 +22,22 @@
 'use strict';
 
 const {
+  ArrayBufferIsView,
   ArrayPrototypeIndexOf,
   NumberIsInteger,
   NumberIsNaN,
   NumberParseInt,
+  JSONParse,
   ObjectDefineProperties,
   ObjectKeys,
   ObjectSetPrototypeOf,
   Promise,
   SafeSet,
   SymbolAsyncIterator,
-  Symbol
+  Symbol,
+  TypedArrayPrototypeSet,
+  TypeError,
+  Uint8Array,
 } = primordials;
 
 module.exports = Readable;
@@ -42,6 +47,8 @@ const EE = require('events');
 const { Stream, prependListener } = require('internal/streams/legacy');
 const { Buffer } = require('buffer');
 
+let Blob;
+
 const {
   addAbortSignal,
 } = require('internal/streams/add-abort-signal');
@@ -58,6 +65,7 @@ const {
 } = require('internal/streams/state');
 
 const {
+  AbortError,
   aggregateTwoErrors,
   codes: {
     ERR_INVALID_ARG_TYPE,
@@ -69,6 +77,8 @@ const {
 const { validateObject } = require('internal/validators');
 
 const kPaused = Symbol('kPaused');
+const kConsume = Symbol('kConsume');
+const kReading = Symbol('kReading');
 
 const { StringDecoder } = require('string_decoder');
 const from = require('internal/streams/from');
@@ -173,6 +183,8 @@ function ReadableState(options, stream, isDuplex) {
 
   this.didRead = false;
 
+  this.dataEmitted = false;
+
   this.decoder = null;
   this.encoding = null;
   if (options && options.encoding) {
@@ -206,6 +218,9 @@ function Readable(options) {
       addAbortSignal(options.signal, this);
   }
 
+  this[kConsume] = null;
+  this[kReading] = false; // Is stream being consumed through Readable API?
+
   Stream.call(this, options);
 
   destroyImpl.construct(this, () => {
@@ -230,6 +245,12 @@ Readable.prototype[EE.captureRejectionSymbol] = function(err) {
 // similar to how Writable.write() returns true if you should
 // write() some more.
 Readable.prototype.push = function(chunk, encoding) {
+  if (this[kConsume] && chunk !== null && !this[kReading]) {
+    // Fast path.
+    encoding ||= this._readableState.defaultEncoding;
+    return consumePush(this[kConsume], chunk, encoding);
+  }
+
   return readableAddChunk(this, chunk, encoding, false);
 };
 
@@ -299,10 +320,12 @@ function readableAddChunk(stream, chunk, encoding, addToFront) {
     maybeReadMore(stream, state);
   }
 
+  const consumed = consumePush(stream[kConsume], chunk, encoding);
+
   // We can push more data if we are below the highWaterMark.
   // Also, if we have no data yet, we can stand some more bytes.
   // This is to work around cases where hwm=0, such as the repl.
-  return !state.ended &&
+  return consumed && !state.ended &&
     (state.length < state.highWaterMark || state.length === 0);
 }
 
@@ -316,6 +339,7 @@ function addChunk(stream, state, chunk, addToFront) {
     } else {
       state.awaitDrainWriters = null;
     }
+    state.dataEmitted = true;
     stream.emit('data', chunk);
   } else {
     // Update the buffer info.
@@ -394,6 +418,27 @@ function howMuchToRead(n, state) {
   return state.ended ? state.length : 0;
 }
 
+
+function _read(self, n) {
+  // Call internal read method
+  try {
+    const result = self._read(n);
+    if (result != null) {
+      const then = result.then;
+      if (typeof then === 'function') {
+        then.call(
+          result,
+          nop,
+          function(err) {
+            errorOrDestroy(self, err);
+          });
+      }
+    }
+  } catch (err) {
+    errorOrDestroy(self, err);
+  }
+}
+
 // You can override either this method, or the async _read(n) below.
 Readable.prototype.read = function(n) {
   debug('read', n);
@@ -488,22 +533,7 @@ Readable.prototype.read = function(n) {
       state.needReadable = true;
 
     // Call internal read method
-    try {
-      const result = this._read(state.highWaterMark);
-      if (result != null) {
-        const then = result.then;
-        if (typeof then === 'function') {
-          then.call(
-            result,
-            nop,
-            function(err) {
-              errorOrDestroy(this, err);
-            });
-        }
-      }
-    } catch (err) {
-      errorOrDestroy(this, err);
-    }
+    _read(this, state.highWaterMark);
 
     state.sync = false;
     // If _read pushed data synchronously, then `reading` will be false,
@@ -541,8 +571,10 @@ Readable.prototype.read = function(n) {
       endReadable(this);
   }
 
-  if (ret !== null)
+  if (ret !== null) {
+    state.dataEmitted = true;
     this.emit('data', ret);
+  }
 
   state.didRead = true;
 
@@ -898,6 +930,8 @@ Readable.prototype.on = function(ev, fn) {
   const state = this._readableState;
 
   if (ev === 'data') {
+    this[kReading] = true;
+
     // Update readableListening so that resume() may be a no-op
     // a few lines down. This is needed to support once('readable').
     state.readableListening = this.listenerCount('readable') > 0;
@@ -906,6 +940,8 @@ Readable.prototype.on = function(ev, fn) {
     if (state.flowing !== false)
       this.resume();
   } else if (ev === 'readable') {
+    this[kReading] = true;
+
     if (!state.endEmitted && !state.readableListening) {
       state.readableListening = state.needReadable = true;
       state.flowing = false;
@@ -1097,6 +1133,22 @@ Readable.prototype.iterator = function(options) {
   return streamToAsyncIterator(this, options);
 };
 
+Readable.prototype.text = function text() {
+  return consume(this, kTextType);
+};
+
+Readable.prototype.json = function json() {
+  return consume(this, kJSONType);
+};
+
+Readable.prototype.blob = function blob() {
+  return consume(this, kBlobType);
+};
+
+Readable.prototype.arrayBuffer = function arrayBuffer() {
+  return consume(this, kArrayBufferType);
+};
+
 function streamToAsyncIterator(stream, options) {
   if (typeof stream.read !== 'function') {
     // v1 stream
@@ -1266,8 +1318,30 @@ ObjectDefineProperties(Readable.prototype, {
     }
   },
 
+  // https://fetch.spec.whatwg.org/#dom-body-bodyused
+  bodyUsed: {
+    get() {
+      return isDisturbed(this);
+    }
+  },
+
+  body: {
+    get() {
+      if (this[kConsume]?.type === kWebStreamType) {
+        return this[kConsume].stream;
+      }
+
+      return consume(this, kWebStreamType);
+    }
+  }
 });
 
+const kWebStreamType = 1;
+const kTextType = 2;
+const kBlobType = 3;
+const kArrayBufferType = 4;
+const kJSONType = 5;
+
 ObjectDefineProperties(ReadableState.prototype, {
   // Legacy getter for `pipesCount`.
   pipesCount: {
@@ -1287,6 +1361,234 @@ ObjectDefineProperties(ReadableState.prototype, {
   }
 });
 
+function isLocked(self) {
+  return self[kConsume]?.stream?.locked === true;
+}
+
+// https://streams.spec.whatwg.org/#readablestream-disturbed
+function isDisturbed(self) {
+  const { _readableState: state } = self;
+  return !!(
+    state.dataEmitted ||
+    state.endEmitted ||
+    state.errorEmitted ||
+    state.closeEmitted
+  );
+}
+
+// https://fetch.spec.whatwg.org/#body-unusable
+function isUnusable(self) {
+  return isDisturbed(self) || isLocked(self);
+}
+
+async function consume(parent, type) {
+  if (isUnusable(parent)) {
+    // eslint-disable-next-line no-restricted-syntax
+    throw new TypeError('unusable');
+  }
+
+  if (parent[kConsume]) {
+    // TODO: Should multiple consume in same tick be possible?
+    // eslint-disable-next-line no-restricted-syntax
+    throw new TypeError('unusable');
+  }
+
+  if (type === kWebStreamType) {
+    const consume = parent[kConsume] = {
+      type,
+      // TODO: Optimized implementation for web streams.
+      stream: lazyWebStreams()
+        .newReadableStreamFromStreamReadable(parent)
+    };
+
+    return consume.stream;
+  }
+
+  return new Promise((resolve, reject) => {
+    parent[kConsume] = {
+      type,
+      parent,
+      resolve,
+      reject,
+      length: 0,
+      decoder: undefined,
+      body: undefined,
+      reading: false,
+      pushed: false,
+      ended: false,
+    };
+
+    parent
+      .once('error', function(err) {
+        consumeFinish(this[kConsume], err);
+      })
+      .once('close', function() {
+        if (this[kConsume].body !== null) {
+          consumeFinish(this[kConsume], new AbortError());
+        }
+      });
+
+    process.nextTick(consumeStart, parent[kConsume]);
+  });
+}
+
+function consumeStart(consume) {
+  if (consume.body === null) {
+    return;
+  }
+
+  const { _readableState: state } = consume.parent;
+
+  for (const chunk of state.buffer) {
+    consumePush(consume, chunk);
+  }
+
+  if (state.endEmitted) {
+    consumeEnd(this[kConsume]);
+  } else {
+    consume.parent.once('end', function() {
+      consumeEnd(this[kConsume]);
+    });
+  }
+
+  if (consume.parent.isPaused()) {
+    consume.parent.resume();
+  }
+
+  while (consume.parent.read() != null);
+}
+
+function consumeEnd(consume) {
+  const { type, body, resolve, decoder, parent, length } = consume;
+
+  try {
+    if (type === kTextType) {
+      resolve(body + (decoder ? decoder.end() : ''));
+    } else if (type === kJSONType) {
+      resolve(JSONParse(body + (decoder ? decoder.end() : '')));
+    } else if (type === kArrayBufferType) {
+      const dst = new Uint8Array(length);
+
+      let pos = 0;
+      for (const buf of body) {
+        TypedArrayPrototypeSet(dst, buf, pos);
+        pos += buf.byteLength;
+      }
+
+      resolve(dst);
+    } else if (type === kBlobType) {
+      if (!Blob) {
+        Blob = require('buffer').Blob;
+      }
+      resolve(new Blob(body));
+    }
+
+    consumeFinish(consume);
+  } catch (err) {
+    parent.destroy(err);
+  }
+}
+
+function consumePush(consume, chunk, encoding) {
+  if (!consume) {
+    return true;
+  }
+
+  if (consume.ended) {
+    return false;
+  }
+
+  if (chunk === null) {
+    consume.ended = true;
+    consume.parent.read();
+    return false;
+  }
+
+  consume.pushed = true;
+
+  if (consume.type === kTextType || consume.type === kJSONType) {
+    consumePushString(consume, chunk, encoding);
+  } else {
+    consumePushBuffer(consume, chunk, encoding);
+  }
+
+  if (!consume.parent[kReading] && !consume.reading) {
+    consume.reading = true;
+    process.nextTick(consumeReadMore, consume);
+  }
+
+  return true;
+}
+
+function consumePushString(consume, chunk, encoding) {
+  if (!consume.body) {
+    consume.body = '';
+  }
+
+  if (typeof chunk === 'string') {
+    if (consume.decoder) {
+      chunk = consume.decoder.write(Buffer.from(chunk, encoding));
+    } else if (encoding !== 'utf8') {
+      chunk = Buffer.from(chunk, encoding).toString();
+    }
+  } else if (ArrayBufferIsView(chunk)) {
+    if (!consume.decoder) {
+      consume.decoder = new StringDecoder('utf8');
+    }
+    encoding = consume.decoder.write(chunk);
+  } else {
+    throw new ERR_INVALID_ARG_TYPE(
+      'chunk', ['Buffer', 'string', 'TypedArray'], chunk);
+  }
+
+  consume.length += chunk.length;
+  consume.body += chunk;
+}
+
+function consumePushBuffer(consume, chunk, encoding) {
+  if (!consume.body) {
+    consume.body = [];
+  }
+
+  if (typeof chunk === 'string') {
+    chunk = Buffer.from(chunk, encoding);
+  } else if (!ArrayBufferIsView(chunk)) {
+    throw new ERR_INVALID_ARG_TYPE(
+      'chunk', ['Buffer', 'string', 'TypedArray'], chunk);
+  }
+
+  consume.length += chunk.byteLength;
+  consume.body.push(chunk);
+}
+
+function consumeReadMore(consume) {
+  if (consume.parent[kReading]) {
+    consume.reading = false;
+    return;
+  }
+
+  consume.pushed = true;
+  while (consume.pushed) {
+    consume.pushed = false;
+    _read(consume.parent);
+  }
+
+  consume.reading = false;
+}
+
+function consumeFinish(consume, err) {
+  if (err) {
+    consume.reject(err);
+  } else {
+    consume.resolve();
+  }
+
+  consume.reject = null;
+  consume.resolve = null;
+  consume.decoder = null;
+  consume.body = null;
+}
+
 // Exposed for testing purposes only.
 Readable._fromList = fromList;
 
diff --git a/test/parallel/test-stream-readable-body-mixin.js b/test/parallel/test-stream-readable-body-mixin.js
new file mode 100644
index 00000000000000..9c4f8bc4f046d7
--- /dev/null
+++ b/test/parallel/test-stream-readable-body-mixin.js
@@ -0,0 +1,355 @@
+'use strict';
+const common = require('../common');
+const assert = require('assert');
+const Readable = require('stream').Readable;
+
+{
+  const r = new Readable({
+    read() {
+      this.push('asd');
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('end', common.mustCall());
+  r.on('close', common.mustCall());
+  r.text().then(common.mustCall((val) => {
+    assert.strictEqual(r.bodyUsed, true);
+    assert.strictEqual(val, 'asd');
+  }));
+  process.nextTick(() => {
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const n = ['a', 's', 'd', null];
+  const r = new Readable({
+    read() {
+      this.push(n.shift());
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('end', common.mustCall());
+  r.on('close', common.mustCall());
+  r.text().then(common.mustCall((val) => {
+    assert.strictEqual(r.bodyUsed, true);
+    assert.strictEqual(val, 'asd');
+  }));
+  process.nextTick(() => {
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const r = new Readable({
+    read() {
+      this.push('asd');
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('error', common.mustCall(() => {
+    assert.strictEqual(r.bodyUsed, true);
+  }));
+  r.on('close', common.mustCall());
+  const _err = new Error();
+  r.text().catch(common.mustCall((err) => assert.strictEqual(err, _err)));
+  r.destroy(_err);
+  process.nextTick(() => {
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const obj = { asd: '123' };
+  const r = new Readable({
+    read() {
+      this.push(JSON.stringify(obj));
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('end', common.mustCall());
+  r.on('close', common.mustCall());
+  r.json().then(common.mustCall((val) => {
+    assert.strictEqual(r.bodyUsed, true);
+    assert.deepStrictEqual(val, obj);
+  }));
+  process.nextTick(() => {
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const r = new Readable({
+    read() {
+      this.push('asd');
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('error', common.mustCall(() => {
+    assert.strictEqual(r.bodyUsed, true);
+  }));
+  r.on('close', common.mustCall());
+  r.json()
+    .catch(common.mustCall((err) => {
+      assert.strictEqual(err.message,
+                         'Unexpected token a in JSON at position 0');
+      assert.strictEqual(r.bodyUsed, true);
+    }));
+  process.nextTick(() => {
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const r = new Readable({
+    read() {
+      this.push('asd');
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('error', common.mustCall(() => {
+    assert.strictEqual(r.bodyUsed, true);
+  }));
+  r.on('close', common.mustCall());
+  r.json().catch(common.mustCall((err) => {
+    assert.strictEqual(r.bodyUsed, true);
+    assert(err);
+  }));
+  process.nextTick(() => {
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const buf = Uint8Array.from('asd');
+  const r = new Readable({
+    read() {
+      this.push(buf);
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('end', common.mustCall());
+  r.on('close', common.mustCall());
+  r.arrayBuffer()
+    .then(common.mustCall((val) => assert.deepStrictEqual(val, buf)));
+  process.nextTick(() => {
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const r = new Readable({
+    read() {
+      this.push('asd');
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.pause();
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('data', common.mustCall(() => {
+    assert.strictEqual(r.bodyUsed, true);
+  }));
+  r.on('end', common.mustCall());
+  r.on('close', common.mustCall());
+  r.text().then(common.mustCall((val) => assert.strictEqual(val, 'asd')));
+  process.nextTick(() => {
+    assert.strictEqual(r.bodyUsed, true);
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const n = ['a', 's', 'd', null];
+  const r = new Readable({
+    read() {
+      this.push(n.shift());
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.pause();
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('data', common.mustCallAtLeast(3));
+  r.on('end', common.mustCall());
+  r.on('close', common.mustCall());
+  r.text().then(common.mustCall((val) => assert.strictEqual(val, 'asd')));
+  process.nextTick(() => {
+    assert.strictEqual(r.bodyUsed, true);
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const r = new Readable({
+    read() {
+      this.push('asd');
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.pause();
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('data', common.mustNotCall());
+  r.on('error', common.mustCall());
+  r.on('close', common.mustCall());
+  const _err = new Error();
+  r.text().catch(common.mustCall((err) => assert.strictEqual(err, _err)));
+  r.destroy(_err);
+  process.nextTick(() => {
+    assert.strictEqual(r.bodyUsed, true);
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const obj = { asd: '123' };
+  const r = new Readable({
+    read() {
+      this.push(JSON.stringify(obj));
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.pause();
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('data', common.mustCall(() => {
+    assert.strictEqual(r.bodyUsed, true);
+  }));
+  r.on('end', common.mustCall());
+  r.on('close', common.mustCall());
+  r.json().then(common.mustCall((val) => assert.deepStrictEqual(val, obj)));
+  process.nextTick(() => {
+    assert.strictEqual(r.bodyUsed, true);
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const r = new Readable({
+    read() {
+      this.push('asd');
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.pause();
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('data', common.mustCall(() => {
+    assert.strictEqual(r.bodyUsed, true);
+  }));
+  r.on('error', common.mustCall(() => {
+    assert.strictEqual(r.bodyUsed, true);
+  }));
+  r.on('close', common.mustCall());
+  r.json()
+    .catch(common.mustCall((err) => assert.strictEqual(
+      err.message, 'Unexpected token a in JSON at position 0')));
+  process.nextTick(() => {
+    assert.strictEqual(r.bodyUsed, true);
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const buf = Uint8Array.from('asd');
+  const r = new Readable({
+    read() {
+      this.push(buf);
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  r.pause();
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('data', common.mustCall());
+  r.on('end', common.mustCall());
+  r.on('close', common.mustCall());
+  r.arrayBuffer()
+    .then(common.mustCall((val) => assert.deepStrictEqual(val, buf)));
+  process.nextTick(() => {
+    assert.strictEqual(r.bodyUsed, true);
+    assert.strictEqual(r.isPaused(), false);
+  });
+}
+
+{
+  const buf = Uint8Array.from('asd');
+  const r = new Readable({
+    read() {
+      this.push(buf);
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('data', common.mustCall(() => {
+    assert.strictEqual(r.bodyUsed, true);
+    r.json()
+      .catch(common.mustCall((err) => assert(
+        err instanceof TypeError)));
+    r.arrayBuffer()
+      .catch(common.mustCall((err) => assert(
+        err instanceof TypeError)));
+    r.blob()
+      .catch(common.mustCall((err) => assert(
+        err instanceof TypeError)));
+    r.text()
+      .catch(common.mustCall((err) => assert(
+        err instanceof TypeError)));
+  }));
+  r.on('error', common.mustNotCall());
+  r.on('end', common.mustCall());
+  r.on('close', common.mustCall());
+}
+
+{
+  const r = new Readable({
+    read() {
+      this.push(null);
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  assert.strictEqual(r.bodyUsed, false);
+  r.on('end', common.mustCall(() => {
+    assert.strictEqual(r.bodyUsed, true);
+    r.json()
+      .catch(common.mustCall((err) => assert(
+        err instanceof TypeError)));
+    r.arrayBuffer()
+      .catch(common.mustCall((err) => assert(
+        err instanceof TypeError)));
+    r.blob()
+      .catch(common.mustCall((err) => assert(
+        err instanceof TypeError)));
+    r.text()
+      .catch(common.mustCall((err) => assert(
+        err instanceof TypeError)));
+  }));
+  r.on('error', common.mustNotCall());
+  r.on('data', common.mustNotCall());
+  r.on('close', common.mustCall());
+}
+
+for (const key of ['text', 'json', 'arrayBuffer', 'blob']) {
+  const r = new Readable({
+    read() {
+    }
+  });
+  assert.strictEqual(r.bodyUsed, false);
+  assert.strictEqual(r.bodyUsed, false);
+  r[key]()
+    .catch(common.mustCall((err) => assert.strictEqual(
+      err.name, 'AbortError')));
+  r.destroy();
+  r.on('error', common.mustNotCall());
+  r.on('end', common.mustNotCall());
+  r.on('data', common.mustNotCall());
+  r.on('close', common.mustCall());
+
+}

From 345ad2e43bdb8bbbcd3f2a95fc78348b09ea96d7 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Wed, 28 Jul 2021 09:59:17 +0200
Subject: [PATCH 02/17] fixup: avoid ?.

---
 lib/internal/streams/readable.js | 10 +++++++---
 tmp.js                           | 32 ++++++++++++++++++++++++++++++++
 2 files changed, 39 insertions(+), 3 deletions(-)
 create mode 100644 tmp.js

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index f8060708730a2c..aadce48e221701 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -1327,7 +1327,7 @@ ObjectDefineProperties(Readable.prototype, {
 
   body: {
     get() {
-      if (this[kConsume]?.type === kWebStreamType) {
+      if (this[kConsume] && this[kConsume].type === kWebStreamType) {
         return this[kConsume].stream;
       }
 
@@ -1362,7 +1362,11 @@ ObjectDefineProperties(ReadableState.prototype, {
 });
 
 function isLocked(self) {
-  return self[kConsume]?.stream?.locked === true;
+  return (
+    self[kConsume] &&
+    self[kConsume].stream &&
+    self[kConsume].stream.locked === true
+  );
 }
 
 // https://streams.spec.whatwg.org/#readablestream-disturbed
@@ -1535,7 +1539,7 @@ function consumePushString(consume, chunk, encoding) {
     if (!consume.decoder) {
       consume.decoder = new StringDecoder('utf8');
     }
-    encoding = consume.decoder.write(chunk);
+    chunk = consume.decoder.write(chunk);
   } else {
     throw new ERR_INVALID_ARG_TYPE(
       'chunk', ['Buffer', 'string', 'TypedArray'], chunk);
diff --git a/tmp.js b/tmp.js
new file mode 100644
index 00000000000000..0b42c4c310200f
--- /dev/null
+++ b/tmp.js
@@ -0,0 +1,32 @@
+const { Readable } = require('stream');
+const { Buffer } = require('buffer');
+
+const points = [];
+
+for (let x = 0; x < 10; x++)
+  points[x] = '' + x;
+
+let buf = Buffer.alloc(12);
+let i = 0, offset = 0;
+
+const stream = new Readable({
+  read() {
+    for (; i < points.length; i++) {
+      const str = points[i];
+      if (str.length + offset + 1 >= 12) {
+        this.push(buf.slice(0, offset), 'utf8');
+        buf = Buffer.alloc(12);
+        offset = 0;
+        continue;
+      }
+      buf.write(str, offset, str.length, 'utf8');
+      offset += str.length;
+      buf.write('\n', offset, 1, 'utf8');
+      offset++;
+    }
+    this.push(buf.slice(0, offset), 'utf8');
+    this.push(null);
+  }
+});
+
+stream.pipe(process.stdout)

From 9cac1e64824d6e7212d1fb4158a99a36cb6647b4 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Wed, 28 Jul 2021 10:26:15 +0200
Subject: [PATCH 03/17] Update lib/internal/streams/readable.js

Co-authored-by: Mestery <mestery@pm.me>
---
 lib/internal/streams/readable.js | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index aadce48e221701..821296a397823c 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -247,8 +247,11 @@ Readable.prototype[EE.captureRejectionSymbol] = function(err) {
 Readable.prototype.push = function(chunk, encoding) {
   if (this[kConsume] && chunk !== null && !this[kReading]) {
     // Fast path.
-    encoding ||= this._readableState.defaultEncoding;
-    return consumePush(this[kConsume], chunk, encoding);
+    return consumePush(
+      this[kConsume],
+      chunk,
+      encoding || this._readableState.defaultEncoding
+    );
   }
 
   return readableAddChunk(this, chunk, encoding, false);

From 74afe254070af861b12387ccb9826daf323a0ce6 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Wed, 28 Jul 2021 10:26:24 +0200
Subject: [PATCH 04/17] fixup

---
 tmp.js | 32 --------------------------------
 1 file changed, 32 deletions(-)
 delete mode 100644 tmp.js

diff --git a/tmp.js b/tmp.js
deleted file mode 100644
index 0b42c4c310200f..00000000000000
--- a/tmp.js
+++ /dev/null
@@ -1,32 +0,0 @@
-const { Readable } = require('stream');
-const { Buffer } = require('buffer');
-
-const points = [];
-
-for (let x = 0; x < 10; x++)
-  points[x] = '' + x;
-
-let buf = Buffer.alloc(12);
-let i = 0, offset = 0;
-
-const stream = new Readable({
-  read() {
-    for (; i < points.length; i++) {
-      const str = points[i];
-      if (str.length + offset + 1 >= 12) {
-        this.push(buf.slice(0, offset), 'utf8');
-        buf = Buffer.alloc(12);
-        offset = 0;
-        continue;
-      }
-      buf.write(str, offset, str.length, 'utf8');
-      offset += str.length;
-      buf.write('\n', offset, 1, 'utf8');
-      offset++;
-    }
-    this.push(buf.slice(0, offset), 'utf8');
-    this.push(null);
-  }
-});
-
-stream.pipe(process.stdout)

From 68f80d900f2072ae3b0d366f1378905a4585679d Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Wed, 28 Jul 2021 10:43:53 +0200
Subject: [PATCH 05/17] fixup: mark as experimental

---
 doc/api/stream.md | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index e4a3db97b1b0d8..ce1e7289495bc5 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -993,6 +993,8 @@ called and `readableFlowing` is not `true`.
 added: REPLACEME
 --->
 
+> Stability: 1 - Experimental
+
 * Returns: {Promise}
 
 Returns a promise resolving to the full content as an array buffer.
@@ -1002,6 +1004,8 @@ Returns a promise resolving to the full content as an array buffer.
 added: REPLACEME
 --->
 
+> Stability: 1 - Experimental
+
 * Returns: {Promise}
 
 Returns a promise resolving to the full content read as
@@ -1012,6 +1016,8 @@ Blob.
 added: REPLACEME
 --->
 
+> Stability: 1 - Experimental
+
 * Returns: {ReadableStream}
 
 Returns a `ReadableStream`.
@@ -1021,6 +1027,8 @@ Returns a `ReadableStream`.
 added: REPLACEME
 --->
 
+> Stability: 1 - Experimental
+
 * Returns: {Boolean}
 
 Returns true if stream has emitted `'data'`, `'end'`, `'error'` or
@@ -1085,6 +1093,8 @@ readable.isPaused(); // === false
 added: REPLACEME
 --->
 
+> Stability: 1 - Experimental
+
 * Returns: {Promise}
 
 Returns a promise resolving to the full content read as
@@ -1396,6 +1406,8 @@ readable.on('data', (chunk) => {
 added: REPLACEME
 --->
 
+> Stability: 1 - Experimental
+
 * Returns: {Promise}
 
 Returns a promise resolving to the full content read as

From 13bbb010dafb23d91eba568b521f15cb9b5297d3 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Wed, 28 Jul 2021 23:07:03 +0200
Subject: [PATCH 06/17] Apply suggestions from code review

Co-authored-by: James M Snell <jasnell@gmail.com>
---
 doc/api/stream.md | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index ce1e7289495bc5..b83464c520981d 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -997,7 +997,7 @@ added: REPLACEME
 
 * Returns: {Promise}
 
-Returns a promise resolving to the full content as an array buffer.
+Returns a promise fulfilled with the full content as an {ArrayBuffer}.
 
 ##### `readable.blob()`
 <!--- YAML>
@@ -1008,8 +1008,7 @@ added: REPLACEME
 
 * Returns: {Promise}
 
-Returns a promise resolving to the full content read as
-Blob.
+Returns a promise fulfilled with the full content as a {Blob}.
 
 ##### `readable.body`
 <!--- YAML>
@@ -1097,8 +1096,8 @@ added: REPLACEME
 
 * Returns: {Promise}
 
-Returns a promise resolving to the full content read as
-utf8 string and parsed to JSON object.
+Returns a promise fulfulled with the full content read as an UTF-8
+string and parsed as a JSON object.
 
 ##### `readable.pause()`
 <!-- YAML
@@ -1410,8 +1409,7 @@ added: REPLACEME
 
 * Returns: {Promise}
 
-Returns a promise resolving to the full content read as
-utf8 string.
+Returns a promise fulfilled with the full content read as an UTF-8 string.
 
 ##### `readable.unpipe([destination])`
 <!-- YAML

From 3c62aa93f2a97067f2bef94cc7c7d8a2d8a53ad2 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 29 Jul 2021 08:39:30 +0200
Subject: [PATCH 07/17] fixup

---
 doc/api/stream.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index b83464c520981d..db8554cc439f9a 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -1096,7 +1096,7 @@ added: REPLACEME
 
 * Returns: {Promise}
 
-Returns a promise fulfulled with the full content read as an UTF-8
+Returns a promise fulfilled with the full content read as an UTF-8
 string and parsed as a JSON object.
 
 ##### `readable.pause()`

From 3c5c7b7afb7e2af634a736d9ce251325f37aad30 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 29 Jul 2021 08:58:34 +0200
Subject: [PATCH 08/17] fixup

---
 doc/api/stream.md | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index db8554cc439f9a..fa380a4d506c63 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -814,8 +814,6 @@ added: v0.9.4
 
 <!--type=class-->
 
-`Readable` implements the body mixin API from the [fetch specification][].
-
 ##### Event: `'close'`
 <!-- YAML
 added: v0.9.4
@@ -3573,7 +3571,6 @@ contain multi-byte characters.
 [child process stdin]: child_process.md#child_process_subprocess_stdin
 [child process stdout and stderr]: child_process.md#child_process_subprocess_stdout
 [crypto]: crypto.md
-[fetch specification]: https://fetch.spec.whatwg.org/#body-mixin
 [fs read streams]: fs.md#fs_class_fs_readstream
 [fs write streams]: fs.md#fs_class_fs_writestream
 [http-incoming-message]: http.md#http_class_http_incomingmessage

From 3eb492f276704b7eadf28a4fe3c86b16a3b7c434 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 29 Jul 2021 09:08:53 +0200
Subject: [PATCH 09/17] fixup: TODO

---
 lib/internal/streams/readable.js | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index 821296a397823c..fe29c74ab30d51 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -1544,6 +1544,8 @@ function consumePushString(consume, chunk, encoding) {
     }
     chunk = consume.decoder.write(chunk);
   } else {
+    // TODO: What if objectMode? Should we just fail consume
+    // or throw?
     throw new ERR_INVALID_ARG_TYPE(
       'chunk', ['Buffer', 'string', 'TypedArray'], chunk);
   }
@@ -1560,6 +1562,8 @@ function consumePushBuffer(consume, chunk, encoding) {
   if (typeof chunk === 'string') {
     chunk = Buffer.from(chunk, encoding);
   } else if (!ArrayBufferIsView(chunk)) {
+    // TODO: What if objectMode? Should we just fail consume
+    // or throw?
     throw new ERR_INVALID_ARG_TYPE(
       'chunk', ['Buffer', 'string', 'TypedArray'], chunk);
   }

From c268c2644c3ba8bcf28a4cadb4a033087e49d9f7 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 29 Jul 2021 09:10:25 +0200
Subject: [PATCH 10/17] fixup: add spec links

---
 lib/internal/streams/readable.js | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index fe29c74ab30d51..afb0e15a1df8e2 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -1136,18 +1136,22 @@ Readable.prototype.iterator = function(options) {
   return streamToAsyncIterator(this, options);
 };
 
+// https://fetch.spec.whatwg.org/#dom-body-text
 Readable.prototype.text = function text() {
   return consume(this, kTextType);
 };
 
+// https://fetch.spec.whatwg.org/#dom-body-json
 Readable.prototype.json = function json() {
   return consume(this, kJSONType);
 };
 
+// https://fetch.spec.whatwg.org/#dom-body-blob
 Readable.prototype.blob = function blob() {
   return consume(this, kBlobType);
 };
 
+// https://fetch.spec.whatwg.org/#dom-body-arraybuffer
 Readable.prototype.arrayBuffer = function arrayBuffer() {
   return consume(this, kArrayBufferType);
 };
@@ -1328,6 +1332,7 @@ ObjectDefineProperties(Readable.prototype, {
     }
   },
 
+  // https://fetch.spec.whatwg.org/#dom-body-body
   body: {
     get() {
       if (this[kConsume] && this[kConsume].type === kWebStreamType) {

From 14798b34b50776edf5d337629385698ae0bedf86 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 29 Jul 2021 09:36:18 +0200
Subject: [PATCH 11/17] Revert "fixup"

This reverts commit 28842033e42c7e72a1c7c82b741dfc23b04ac3e1.
---
 doc/api/stream.md | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index fa380a4d506c63..db8554cc439f9a 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -814,6 +814,8 @@ added: v0.9.4
 
 <!--type=class-->
 
+`Readable` implements the body mixin API from the [fetch specification][].
+
 ##### Event: `'close'`
 <!-- YAML
 added: v0.9.4
@@ -3571,6 +3573,7 @@ contain multi-byte characters.
 [child process stdin]: child_process.md#child_process_subprocess_stdin
 [child process stdout and stderr]: child_process.md#child_process_subprocess_stdout
 [crypto]: crypto.md
+[fetch specification]: https://fetch.spec.whatwg.org/#body-mixin
 [fs read streams]: fs.md#fs_class_fs_readstream
 [fs write streams]: fs.md#fs_class_fs_writestream
 [http-incoming-message]: http.md#http_class_http_incomingmessage

From 24ae8d2e9e33601c696b02bf107b87d738ce0d10 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 29 Jul 2021 13:07:51 +0200
Subject: [PATCH 12/17] fixup: yaml

---
 doc/api/stream.md | 12 ++++++------
 1 file changed, 6 insertions(+), 6 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index db8554cc439f9a..f1eb774bab9c49 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -989,7 +989,7 @@ The `'resume'` event is emitted when [`stream.resume()`][stream-resume] is
 called and `readableFlowing` is not `true`.
 
 ##### `readable.arrayBuffer()`
-<!--- YAML>
+<!-- YAML>
 added: REPLACEME
 --->
 
@@ -1000,7 +1000,7 @@ added: REPLACEME
 Returns a promise fulfilled with the full content as an {ArrayBuffer}.
 
 ##### `readable.blob()`
-<!--- YAML>
+<!-- YAML>
 added: REPLACEME
 --->
 
@@ -1011,7 +1011,7 @@ added: REPLACEME
 Returns a promise fulfilled with the full content as a {Blob}.
 
 ##### `readable.body`
-<!--- YAML>
+<!-- YAML>
 added: REPLACEME
 --->
 
@@ -1022,7 +1022,7 @@ added: REPLACEME
 Returns a `ReadableStream`.
 
 ##### `readable.bodyUsed`
-<!--- YAML>
+<!-- YAML>
 added: REPLACEME
 --->
 
@@ -1088,7 +1088,7 @@ readable.isPaused(); // === false
 ```
 
 ##### `readable.json()`
-<!--- YAML>
+<!-- YAML>
 added: REPLACEME
 --->
 
@@ -1401,7 +1401,7 @@ readable.on('data', (chunk) => {
 ```
 
 ##### `readable.text()`
-<!--- YAML>
+<!-- YAML>
 added: REPLACEME
 --->
 

From 2e9fa0dd1ffb706bcddf44c51f1bd1a82ae7efb6 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 29 Jul 2021 13:26:42 +0200
Subject: [PATCH 13/17] fixup

---
 doc/api/stream.md | 14 +++++++-------
 1 file changed, 7 insertions(+), 7 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index f1eb774bab9c49..eb3eef791d2104 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -991,7 +991,7 @@ called and `readableFlowing` is not `true`.
 ##### `readable.arrayBuffer()`
 <!-- YAML>
 added: REPLACEME
---->
+-->
 
 > Stability: 1 - Experimental
 
@@ -1002,7 +1002,7 @@ Returns a promise fulfilled with the full content as an {ArrayBuffer}.
 ##### `readable.blob()`
 <!-- YAML>
 added: REPLACEME
---->
+-->
 
 > Stability: 1 - Experimental
 
@@ -1013,7 +1013,7 @@ Returns a promise fulfilled with the full content as a {Blob}.
 ##### `readable.body`
 <!-- YAML>
 added: REPLACEME
---->
+-->
 
 > Stability: 1 - Experimental
 
@@ -1022,9 +1022,9 @@ added: REPLACEME
 Returns a `ReadableStream`.
 
 ##### `readable.bodyUsed`
-<!-- YAML>
+<!-- YAML
 added: REPLACEME
---->
+-->
 
 > Stability: 1 - Experimental
 
@@ -1090,7 +1090,7 @@ readable.isPaused(); // === false
 ##### `readable.json()`
 <!-- YAML>
 added: REPLACEME
---->
+-->
 
 > Stability: 1 - Experimental
 
@@ -1403,7 +1403,7 @@ readable.on('data', (chunk) => {
 ##### `readable.text()`
 <!-- YAML>
 added: REPLACEME
---->
+-->
 
 > Stability: 1 - Experimental
 

From 72588db13b928df3d01a93d8fb0627d1bc2a6dc0 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 29 Jul 2021 13:55:18 +0200
Subject: [PATCH 14/17] fixup

---
 doc/api/stream.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index eb3eef791d2104..67a7618f7f2e60 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -1028,7 +1028,7 @@ added: REPLACEME
 
 > Stability: 1 - Experimental
 
-* Returns: {Boolean}
+* Returns: {boolean}
 
 Returns true if stream has emitted `'data'`, `'end'`, `'error'` or
 `'close'`.

From 95a30f08f393672adeeac993038c843d94934200 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 29 Jul 2021 14:20:45 +0200
Subject: [PATCH 15/17] fixup

---
 doc/api/stream.md | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/doc/api/stream.md b/doc/api/stream.md
index 67a7618f7f2e60..c9bff3bab9e41a 100644
--- a/doc/api/stream.md
+++ b/doc/api/stream.md
@@ -989,7 +989,7 @@ The `'resume'` event is emitted when [`stream.resume()`][stream-resume] is
 called and `readableFlowing` is not `true`.
 
 ##### `readable.arrayBuffer()`
-<!-- YAML>
+<!-- YAML
 added: REPLACEME
 -->
 
@@ -1000,7 +1000,7 @@ added: REPLACEME
 Returns a promise fulfilled with the full content as an {ArrayBuffer}.
 
 ##### `readable.blob()`
-<!-- YAML>
+<!-- YAML
 added: REPLACEME
 -->
 
@@ -1011,7 +1011,7 @@ added: REPLACEME
 Returns a promise fulfilled with the full content as a {Blob}.
 
 ##### `readable.body`
-<!-- YAML>
+<!-- YAML
 added: REPLACEME
 -->
 
@@ -1088,7 +1088,7 @@ readable.isPaused(); // === false
 ```
 
 ##### `readable.json()`
-<!-- YAML>
+<!-- YAML
 added: REPLACEME
 -->
 
@@ -1401,7 +1401,7 @@ readable.on('data', (chunk) => {
 ```
 
 ##### `readable.text()`
-<!-- YAML>
+<!-- YAML
 added: REPLACEME
 -->
 

From d3df1ece7b6ba2e32361e00da6ee4c662a1fa2a1 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Thu, 29 Jul 2021 15:01:23 +0200
Subject: [PATCH 16/17] fixuP

---
 .../streams/readable-readstring-async.js      | 37 -------------------
 benchmark/streams/readable-readstring-data.js | 32 ----------------
 .../streams/readable-readstring-native.js     | 25 -------------
 benchmark/streams/readable-readstring-text.js | 28 --------------
 test/parallel/test-http-unix-socket.js        |  6 +--
 5 files changed, 3 insertions(+), 125 deletions(-)
 delete mode 100644 benchmark/streams/readable-readstring-async.js
 delete mode 100644 benchmark/streams/readable-readstring-data.js
 delete mode 100644 benchmark/streams/readable-readstring-native.js
 delete mode 100644 benchmark/streams/readable-readstring-text.js

diff --git a/benchmark/streams/readable-readstring-async.js b/benchmark/streams/readable-readstring-async.js
deleted file mode 100644
index f2eff78ba575f2..00000000000000
--- a/benchmark/streams/readable-readstring-async.js
+++ /dev/null
@@ -1,37 +0,0 @@
-'use strict';
-
-const common = require('../common');
-const Readable = require('stream').Readable;
-
-const bench = common.createBenchmark(main, {
-  n: [50e2]
-});
-
-function main({ n }) {
-  const b = 'some random string';
-  const s = new Readable();
-  function noop() {}
-  s._read = noop;
-
-  bench.start();
-
-  text(s).then((ret) => {
-    console.log(ret.length);
-    bench.end(n, ret);
-  });
-
-  for (let k = 0; k < n; ++k) {
-    for (let i = 0; i < 1e3; ++i)
-      s.push(b);
-  }
-  s.push(null);
-}
-
-async function text(s) {
-  s.setEncoding('utf8');
-  let ret = '';
-  for await (const chunk of s) {
-    ret += chunk;
-  }
-  return ret;
-}
diff --git a/benchmark/streams/readable-readstring-data.js b/benchmark/streams/readable-readstring-data.js
deleted file mode 100644
index 03d9e36d38befc..00000000000000
--- a/benchmark/streams/readable-readstring-data.js
+++ /dev/null
@@ -1,32 +0,0 @@
-'use strict';
-
-const common = require('../common');
-const Readable = require('stream').Readable;
-
-const bench = common.createBenchmark(main, {
-  n: [50e2]
-});
-
-function main({ n }) {
-  const b = 'some random string';
-  const s = new Readable();
-  function noop() {}
-  s._read = noop;
-
-  bench.start();
-
-  let ret = '';
-  s.setEncoding('utf8');
-  s.on('data', (chunk) => {
-    ret += chunk;
-  }).on('end', () => {
-    console.log(ret.length);
-    bench.end(n, ret);
-  });
-
-  for (let k = 0; k < n; ++k) {
-    for (let i = 0; i < 1e3; ++i)
-      s.push(b);
-  }
-  s.push(null);
-}
diff --git a/benchmark/streams/readable-readstring-native.js b/benchmark/streams/readable-readstring-native.js
deleted file mode 100644
index 1374bcf53aff76..00000000000000
--- a/benchmark/streams/readable-readstring-native.js
+++ /dev/null
@@ -1,25 +0,0 @@
-'use strict';
-
-const common = require('../common');
-const Readable = require('stream').Readable;
-
-const bench = common.createBenchmark(main, {
-  n: [50e2]
-});
-
-function main({ n }) {
-  const b = 'some random string';
-  const s = new Readable();
-  function noop() {}
-  s._read = noop;
-
-  bench.start();
-
-  let ret = '';
-  for (let k = 0; k < n; ++k) {
-    for (let i = 0; i < 1e3; ++i)
-      ret += b;
-  }
-  console.log(ret.length);
-  bench.end(n);
-}
diff --git a/benchmark/streams/readable-readstring-text.js b/benchmark/streams/readable-readstring-text.js
deleted file mode 100644
index eba63f4077646e..00000000000000
--- a/benchmark/streams/readable-readstring-text.js
+++ /dev/null
@@ -1,28 +0,0 @@
-'use strict';
-
-const common = require('../common');
-const Readable = require('stream').Readable;
-
-const bench = common.createBenchmark(main, {
-  n: [50e2]
-});
-
-function main({ n }) {
-  const b = 'some random string';
-  const s = new Readable();
-  function noop() {}
-  s._read = noop;
-
-  bench.start();
-
-  s.text().then((ret) => {
-    console.log(ret.length);
-    bench.end(n);
-  });
-
-  for (let k = 0; k < n; ++k) {
-    for (let i = 0; i < 1e3; ++i)
-      s.push(b);
-  }
-  s.push(null);
-}
diff --git a/test/parallel/test-http-unix-socket.js b/test/parallel/test-http-unix-socket.js
index f8362d6183d8e1..e65c42ff243fc9 100644
--- a/test/parallel/test-http-unix-socket.js
+++ b/test/parallel/test-http-unix-socket.js
@@ -48,15 +48,15 @@ server.listen(common.PIPE, common.mustCall(function() {
     assert.strictEqual(res.statusCode, 200);
     assert.strictEqual(res.headers['content-type'], 'text/plain');
 
-    res.body = '';
+    res._body = '';
     res.setEncoding('utf8');
 
     res.on('data', function(chunk) {
-      res.body += chunk;
+      res._body += chunk;
     });
 
     res.on('end', common.mustCall(function() {
-      assert.strictEqual(res.body, 'hello world\n');
+      assert.strictEqual(res._body, 'hello world\n');
       server.close(common.mustCall(function(error) {
         assert.strictEqual(error, undefined);
         server.close(common.expectsError({

From bb368bf7f4cfa1d219fbe8da820d4d089f691618 Mon Sep 17 00:00:00 2001
From: Robert Nagy <ronagy@icloud.com>
Date: Fri, 30 Jul 2021 09:44:42 +0200
Subject: [PATCH 17/17] fixup: body compat

---
 lib/internal/streams/readable.js                 | 12 ++++++++++++
 test/parallel/test-stream-readable-body-mixin.js | 10 ++++++++++
 2 files changed, 22 insertions(+)

diff --git a/lib/internal/streams/readable.js b/lib/internal/streams/readable.js
index afb0e15a1df8e2..b1a12708a1fed2 100644
--- a/lib/internal/streams/readable.js
+++ b/lib/internal/streams/readable.js
@@ -79,6 +79,8 @@ const { validateObject } = require('internal/validators');
 const kPaused = Symbol('kPaused');
 const kConsume = Symbol('kConsume');
 const kReading = Symbol('kReading');
+const kBody = Symbol('kBody');
+const kHasBody = Symbol('kHasBody');
 
 const { StringDecoder } = require('string_decoder');
 const from = require('internal/streams/from');
@@ -1335,11 +1337,21 @@ ObjectDefineProperties(Readable.prototype, {
   // https://fetch.spec.whatwg.org/#dom-body-body
   body: {
     get() {
+      // Compat
+      if (this[kHasBody]) {
+        return this[kBody];
+      }
+
       if (this[kConsume] && this[kConsume].type === kWebStreamType) {
         return this[kConsume].stream;
       }
 
       return consume(this, kWebStreamType);
+    },
+    // Compat
+    set(value) {
+      this[kBody] = value;
+      this[kHasBody] = true;
     }
   }
 });
diff --git a/test/parallel/test-stream-readable-body-mixin.js b/test/parallel/test-stream-readable-body-mixin.js
index 9c4f8bc4f046d7..7f76921f4107bc 100644
--- a/test/parallel/test-stream-readable-body-mixin.js
+++ b/test/parallel/test-stream-readable-body-mixin.js
@@ -353,3 +353,13 @@ for (const key of ['text', 'json', 'arrayBuffer', 'blob']) {
   r.on('close', common.mustCall());
 
 }
+
+{
+  const r = new Readable({
+    read() {
+      this.push(null);
+    }
+  });
+  r.body = 'asd';
+  assert.strictEqual(r.body, 'asd');
+}