stream: cleanup async handling

Cleanup async stream method handling.

Author: ronag

lib/internal/streams/destroy.js

@@ -70,12 +70,16 @@ function destroy(err, cb) {
 
 function _destroy(self, err, cb) {
   let called = false;
-  const result = self._destroy(err || null, (err) => {
-    const r = self._readableState;
-    const w = self._writableState;
 
+  function onDestroy(err) {
+    if (called) {
+      return;
+    }
     called = true;
 
+    const r = self._readableState;
+    const w = self._writableState;
+
     checkError(err, w, r);
 
     if (w) {
@@ -94,64 +98,24 @@ function _destroy(self, err, cb) {
     } else {
       process.nextTick(emitCloseNT, self);
     }
-  });
-  if (result !== undefined && result !== null) {
-    try {
+  }
+  try {
+    const result = self._destroy(err || null, onDestroy);
+    if (result != null) {
       const then = result.then;
       if (typeof then === 'function') {
         then.call(
           result,
           function() {
-            if (called)
-              return;
-
-            const r = self._readableState;
-            const w = self._writableState;
-
-            if (w) {
-              w.closed = true;
-            }
-            if (r) {
-              r.closed = true;
-            }
-
-            if (typeof cb === 'function') {
-              process.nextTick(cb);
-            }
-
-            process.nextTick(emitCloseNT, self);
+            process.nextTick(onDestroy, null);
           },
           function(err) {
-            const r = self._readableState;
-            const w = self._writableState;
-            err.stack; // eslint-disable-line no-unused-expressions
-
-            called = true;
-
-            if (w && !w.errored) {
-              w.errored = err;
-            }
-            if (r && !r.errored) {
-              r.errored = err;
-            }
-
-            if (w) {
-              w.closed = true;
-            }
-            if (r) {
-              r.closed = true;
-            }
-
-            if (typeof cb === 'function') {
-              process.nextTick(cb, err);
-            }
-
-            process.nextTick(emitErrorCloseNT, self, err);
+            process.nextTick(onDestroy, err);
           });
       }
-    } catch (err) {
-      process.nextTick(emitErrorNT, self, err);
     }
+  } catch (err) {
+    onDestroy(err);
   }
 }
 
@@ -285,74 +249,52 @@ function construct(stream, cb) {
 }
 
 function constructNT(stream) {
-  const r = stream._readableState;
-  const w = stream._writableState;
-  // With duplex streams we use the writable side for state.
-  const s = w || r;
-
   let called = false;
-  const result = stream._construct((err) => {
+
+  function onConstruct(err) {
+    if (called) {
+      errorOrDestroy(stream, err ?? new ERR_MULTIPLE_CALLBACK());
+      return;
+    }
+    called = true;
+
+    const r = stream._readableState;
+    const w = stream._writableState;
+    const s = w || r;
+
     if (r) {
       r.constructed = true;
     }
     if (w) {
       w.constructed = true;
     }
 
-    if (called) {
-      err = new ERR_MULTIPLE_CALLBACK();
-    } else {
-      called = true;
-    }
-
     if (s.destroyed) {
       stream.emit(kDestroy, err);
     } else if (err) {
       errorOrDestroy(stream, err, true);
     } else {
       process.nextTick(emitConstructNT, stream);
     }
-  });
-  if (result !== undefined && result !== null) {
-    try {
+  }
+
+  try {
+    const result = stream._construct(onConstruct);
+    if (result != null) {
       const then = result.then;
       if (typeof then === 'function') {
         then.call(
           result,
           function() {
-            // If the callback was invoked, do nothing further.
-            if (called)
-              return;
-            if (r) {
-              r.constructed = true;
-            }
-            if (w) {
-              w.constructed = true;
-            }
-            if (s.destroyed) {
-              process.nextTick(() => stream.emit(kDestroy));
-            } else {
-              process.nextTick(emitConstructNT, stream);
-            }
+            process.nextTick(onConstruct, null);
           },
           function(err) {
-            if (r) {
-              r.constructed = true;
-            }
-            if (w) {
-              w.constructed = true;
-            }
-            called = true;
-            if (s.destroyed) {
-              process.nextTick(() => stream.emit(kDestroy, err));
-            } else {
-              process.nextTick(errorOrDestroy, stream, err);
-            }
+            process.nextTick(onConstruct, err);
           });
       }
-    } catch (err) {
-      process.nextTick(emitErrorNT, stream, err);
     }
+  } catch (err) {
+    onConstruct(err);
   }
 }
 

lib/internal/streams/readable.js

@@ -483,8 +483,25 @@ Readable.prototype.read = function(n) {
     // If the length is currently zero, then we *need* a readable event.
     if (state.length === 0)
       state.needReadable = true;
+
     // Call internal read method
-    this._read(state.highWaterMark);
+    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);
+    }
+
     state.sync = false;
     // If _read pushed data synchronously, then `reading` will be false,
     // and we need to re-evaluate how much data we can return to the user.

lib/internal/streams/writable.js

@@ -658,9 +658,15 @@ function needFinish(state) {
 }
 
 function callFinal(stream, state) {
-  state.sync = true;
-  state.pendingcb++;
-  const result = stream._final((err) => {
+  let called = false;
+
+  function onFinish(err) {
+    if (called) {
+      errorOrDestroy(stream, err ?? ERR_MULTIPLE_CALLBACK());
+      return;
+    }
+    called = true;
+
     state.pendingcb--;
     if (err) {
       const onfinishCallbacks = state[kOnFinished].splice(0);
@@ -677,33 +683,30 @@ function callFinal(stream, state) {
       state.pendingcb++;
       process.nextTick(finish, stream, state);
     }
-  });
-  if (result !== undefined && result !== null) {
-    try {
+  }
+
+  state.sync = true;
+  state.pendingcb++;
+
+  try {
+    const result = stream._final(onFinish);
+    if (result != null) {
       const then = result.then;
       if (typeof then === 'function') {
         then.call(
           result,
           function() {
-            if (state.prefinished)
-              return;
-            state.prefinish = true;
-            process.nextTick(() => stream.emit('prefinish'));
-            state.pendingcb++;
-            process.nextTick(finish, stream, state);
+            process.nextTick(onFinish, null);
           },
           function(err) {
-            const onfinishCallbacks = state[kOnFinished].splice(0);
-            for (let i = 0; i < onfinishCallbacks.length; i++) {
-              process.nextTick(onfinishCallbacks[i], err);
-            }
-            process.nextTick(errorOrDestroy, stream, err, state.sync);
+            process.nextTick(onFinish, err);
           });
       }
-    } catch (err) {
-      process.nextTick(errorOrDestroy, stream, err, state.sync);
     }
+  } catch (err) {
+    onFinish(stream, state, err);
   }
+
   state.sync = false;
 }