fixup

Author: ronag

lib/_http_incoming.js

@@ -87,7 +87,6 @@ function IncomingMessage(socket) {
   this.statusMessage = null;
   this.client = socket;
 
-  // TODO: Deprecate and remove.
   this._consuming = false;
   // Flag for when we decide that this message cannot possibly be
   // read by the user, so there's no point continuing to handle it.

lib/_http_server.js

@@ -804,7 +804,7 @@ function resOnFinish(req, res, socket, state, server) {
   // If the user never called req.read(), and didn't pipe() or
   // .resume() or .on('data'), then we call req._dump() so that the
   // bytes will be pulled off the wire.
-  if (!req.readableDidRead)
+  if (!req._consuming && !req._readableState.resumeScheduled)
     req._dump();
 
   // Make sure the requestTimeout is cleared before finishing.

lib/internal/streams/compose.js

@@ -2,31 +2,17 @@
 
 const pipeline = require('internal/streams/pipeline');
 const Duplex = require('internal/streams/duplex');
-const { createDeferredPromise } = require('internal/util');
 const { destroyer } = require('internal/streams/destroy');
-const from = require('internal/streams/from');
 const {
-  isIterable,
-  isReadableNodeStream,
-  isWritableNodeStream,
   isNodeStream,
-  isDestroyed,
-  isReadableFinished,
-  isWritableEnded,
 } = require('internal/streams/utils');
-const {
-  PromiseResolve,
-} = primordials;
 const {
   AbortError,
   codes: {
-    ERR_INVALID_ARG_TYPE,
     ERR_INVALID_ARG_VALUE,
-    ERR_INVALID_RETURN_VALUE,
     ERR_MISSING_ARGS,
   },
 } = require('internal/errors');
-const assert = require('internal/assert');
 
 function isReadable(stream) {
   const r = isReadableNodeStream(stream);
@@ -70,18 +56,18 @@ module.exports = function compose(...streams) {
   }
 
   if (streams.length === 1) {
-    return makeDuplex(streams[0], 'streams[0]');
+    return Duplex.from(streams[0]);
   }
 
   const orgStreams = [...streams];
 
   if (typeof streams[0] === 'function') {
-    streams[0] = makeDuplex(streams[0], 'streams[0]');
+    streams[0] = Duplex.from(streams[0]);
   }
 
   if (typeof streams[streams.length - 1] === 'function') {
     const idx = streams.length - 1;
-    streams[idx] = makeDuplex(streams[idx], `streams[${idx}]`);
+    streams[idx] = Duplex.from(streams[idx]);
   }
 
   for (let n = 0; n < streams.length; ++n) {
@@ -134,7 +120,7 @@ module.exports = function compose(...streams) {
   // Implement Writable/Readable/Duplex traits.
   // See, https://github.com/nodejs/node/pull/33515.
   d = new ComposeDuplex({
-    highWaterMark: 1,
+    // TODO (ronag): highWaterMark?
     writableObjectMode: !!head?.writableObjectMode,
     readableObjectMode: !!tail?.writableObjectMode,
     writable,
@@ -220,82 +206,3 @@ module.exports = function compose(...streams) {
 
   return d;
 };
-
-function makeDuplex(stream, name) {
-  let ret;
-  if (typeof stream === 'function') {
-    assert(stream.length > 0);
-
-    const { value, write, final } = fromAsyncGen(stream);
-
-    if (isIterable(value)) {
-      ret = from(ComposeDuplex, value, {
-        objectMode: true,
-        highWaterMark: 1,
-        write,
-        final
-      });
-    } else if (typeof value?.then === 'function') {
-      const promise = PromiseResolve(value)
-        .then((val) => {
-          if (val != null) {
-            throw new ERR_INVALID_RETURN_VALUE('nully', name, val);
-          }
-        })
-        .catch((err) => {
-          destroyer(ret, err);
-        });
-
-      ret = new ComposeDuplex({
-        objectMode: true,
-        highWaterMark: 1,
-        readable: false,
-        write,
-        final(cb) {
-          final(() => promise.then(cb, cb));
-        }
-      });
-    } else {
-      throw new ERR_INVALID_RETURN_VALUE(
-        'Iterable, AsyncIterable or AsyncFunction', name, value);
-    }
-  } else if (isNodeStream(stream)) {
-    ret = stream;
-  } else if (isIterable(stream)) {
-    ret = from(ComposeDuplex, stream, {
-      objectMode: true,
-      highWaterMark: 1,
-      writable: false
-    });
-  } else {
-    throw new ERR_INVALID_ARG_TYPE(
-      name,
-      ['Stream', 'Iterable', 'AsyncIterable', 'Function'],
-      stream)
-    ;
-  }
-  return ret;
-}
-
-function fromAsyncGen(fn) {
-  let { promise, resolve } = createDeferredPromise();
-  const value = fn(async function*() {
-    while (true) {
-      const { chunk, done, cb } = await promise;
-      process.nextTick(cb);
-      if (done) return;
-      yield chunk;
-      ({ promise, resolve } = createDeferredPromise());
-    }
-  }());
-
-  return {
-    value,
-    write(chunk, encoding, cb) {
-      resolve({ chunk, done: false, cb });
-    },
-    final(cb) {
-      resolve({ done: true, cb });
-    }
-  };
-}

lib/internal/streams/readable.js

@@ -527,8 +527,6 @@ Readable.prototype.read = function(n) {
     this.emit('data', ret);
   }
 
-  state.didRead = true;
-
   return ret;
 };
 
@@ -832,9 +830,7 @@ function pipeOnDrain(src, dest) {
 
     if ((!state.awaitDrainWriters || state.awaitDrainWriters.size === 0) &&
       EE.listenerCount(src, 'data')) {
-      // TODO(ronag): Call resume() instead?
       state.flowing = true;
-      state.didRead = true;
       flow(src);
     }
   };
@@ -982,7 +978,6 @@ Readable.prototype.resume = function() {
 function resume(stream, state) {
   if (!state.resumeScheduled) {
     state.resumeScheduled = true;
-    state.didRead = true;
     process.nextTick(resume_, stream, state);
   }
 }