fixup

Author: ronag

lib/internal/streams/operators.js

@@ -48,21 +48,21 @@ module.exports.map = function map(stream, fn, options) {
       callback(err);
     }
   });
+  const state = ret._readableState;
 
   async function read () {
-    if (ret.readableLength && ret.readableLength >= ret.readableHighWaterMark) {
-      return;
-    }
-
     try {
       reading = true;
-      while (queue.length && !ret.destroyed) {
+      while (
+        queue.length &&
+        !state.destroyed &&
+        (!state.length || state.length < state.highWaterMark)
+      ) {
         const [err, val] = await queue.shift();
         if (err) {
           ret.destroy(err);
-        } else if (!ret.push(val)) {
-          break;
         } else {
+          ret.push(val);
           pump();
         }
       }
@@ -82,7 +82,7 @@ module.exports.map = function map(stream, fn, options) {
 
   function enqueue(val) {
     queue.push(val);
-    if (!reading) {
+    if (!reading && state.length < state.highWaterMark) {
       read();
     }
   }
@@ -93,7 +93,6 @@ module.exports.map = function map(stream, fn, options) {
       if (val === null) {
         return;
       }
-
       enqueue(wrap(val));
     }
   }