src: refactor WriteWrap and ShutdownWraps

Encapsulate stream requests more:

- `WriteWrap` and `ShutdownWrap` classes are now tailored to the
  streams on which they are used. In particular, for most streams
  these are now plain `AsyncWrap`s and do not carry the overhead
  of unused libuv request data.
- Provide generic `Write()` and `Shutdown()` methods that wrap
  around the actual implementations, and make *usage* of streams
  easier, rather than implementing; for example, wrap objects
  don’t need to be provided by callers anymore.
- Use `EmitAfterWrite()` and `EmitAfterShutdown()` handlers to
  call the corresponding JS handlers, rather than always trying
  to call them. This makes usage of streams by other C++ code
  easier and leaner.

Also fix up some tests that were previously not actually testing
asynchronicity when the comments indicated that they would.

PR-URL: https://github.com/nodejs/node/pull/18676
Reviewed-By: Ben Noordhuis <info@bnoordhuis.nl>
Reviewed-By: Anatoli Papirovski <apapirovski@mac.com>
Reviewed-By: James M Snell <jasnell@gmail.com>

3 files changed, 14 insertions, 15 deletions

diff --git a/lib/internal/http2/core.js b/lib/internal/http2/core.js
index 2fdcb8d638..dc01fb14e6 100644
--- a/lib/internal/http2/core.js
+++ b/lib/internal/http2/core.js
@@ -1399,20 +1399,19 @@ function trackWriteState(stream, bytes) {
   session[kHandle].chunksSentSinceLastWrite = 0;
 }
 
-function afterDoStreamWrite(status, handle, req) {
+function afterDoStreamWrite(status, handle) {
   const stream = handle[kOwner];
   const session = stream[kSession];
 
   stream[kUpdateTimer]();
 
-  const { bytes } = req;
+  const { bytes } = this;
   stream[kState].writeQueueSize -= bytes;
 
   if (session !== undefined)
     session[kState].writeQueueSize -= bytes;
-  if (typeof req.callback === 'function')
-    req.callback(null);
-  req.handle = undefined;
+  if (typeof this.callback === 'function')
+    this.callback(null);
 }
 
 function streamOnResume() {
diff --git a/lib/internal/wrap_js_stream.js b/lib/internal/wrap_js_stream.js
index aed26ed8c6..feacab267b 100644
--- a/lib/internal/wrap_js_stream.js
+++ b/lib/internal/wrap_js_stream.js
@@ -115,9 +115,9 @@ class JSStreamWrap extends Socket {
 
     const handle = this._handle;
 
-    this.stream.end(() => {
-      // Ensure that write was dispatched
-      setImmediate(() => {
+    setImmediate(() => {
+      // Ensure that write is dispatched asynchronously.
+      this.stream.end(() => {
         this.finishShutdown(handle, 0);
       });
     });
diff --git a/lib/net.js b/lib/net.js
index 48dd690e89..1bc28e856d 100644
--- a/lib/net.js
+++ b/lib/net.js
@@ -335,7 +335,7 @@ function onSocketFinish() {
 }
 
 
-function afterShutdown(status, handle, req) {
+function afterShutdown(status, handle) {
   var self = handle.owner;
 
   debug('afterShutdown destroyed=%j', self.destroyed,
@@ -869,12 +869,12 @@ protoGetter('bytesWritten', function bytesWritten() {
 });
 
 
-function afterWrite(status, handle, req, err) {
+function afterWrite(status, handle, err) {
   var self = handle.owner;
   if (self !== process.stderr && self !== process.stdout)
     debug('afterWrite', status);
 
-  if (req.async)
+  if (this.async)
     self[kLastWriteQueueSize] = 0;
 
   // callback may come after call to destroy.
@@ -884,9 +884,9 @@ function afterWrite(status, handle, req, err) {
   }
 
   if (status < 0) {
-    var ex = errnoException(status, 'write', req.error);
+    var ex = errnoException(status, 'write', this.error);
     debug('write failure', ex);
-    self.destroy(ex, req.cb);
+    self.destroy(ex, this.cb);
     return;
   }
 
@@ -895,8 +895,8 @@ function afterWrite(status, handle, req, err) {
   if (self !== process.stderr && self !== process.stdout)
     debug('afterWrite call cb');
 
-  if (req.cb)
-    req.cb.call(undefined);
+  if (this.cb)
+    this.cb.call(undefined);
 }