fix: stream body handling

nodejs · Nov 1, 2023 · 4c8740b · 4c8740b
1 parent 41c253d
commit 4c8740b
Show file tree

Hide file tree

Showing 2 changed files with 45 additions and 20 deletions.
diff --git a/lib/client.js b/lib/client.js
@@ -1462,23 +1462,7 @@ function _resume (client, sync) {
       return
     }
 
-    if (util.isStream(request.body) && util.bodyLength(request.body) === 0) {
-      request.body
-        .on('data', /* istanbul ignore next */ function () {
-          /* istanbul ignore next */
-          assert(false)
-        })
-        .on('error', function (err) {
-          errorRequest(client, request, err)
-        })
-        .on('end', function () {
-          util.destroy(this)
-        })
-
-      request.body = null
-    }
-
-    if (client[kRunning] > 0 &&
+    if (client[kRunning] > 0 && util.bodyLength(request.body) !== 0 &&
       (util.isStream(request.body) || util.isAsyncIterable(request.body))) {
       // Request with stream or iterator body can error while other requests
       // are inflight and indirectly error those as well.
@@ -1527,7 +1511,9 @@ function write (client, request) {
     body.read(0)
   }
 
-  let contentLength = util.bodyLength(body)
+  const bodyLength = util.bodyLength(body)
+
+  let contentLength = bodyLength
 
   if (contentLength === null) {
     contentLength = request.contentLength
@@ -1623,7 +1609,7 @@ function write (client, request) {
   }
 
   /* istanbul ignore else: assertion */
-  if (!body) {
+  if (!body || bodyLength === 0) {
     if (contentLength === 0) {
       socket.write(`${header}content-length: 0\r\n\r\n`, 'latin1')
     } else {

diff --git a/lib/core/request.js b/lib/core/request.js
@@ -112,9 +112,33 @@ class Request {
 
     this.method = method
 
+    this.abort = null
+
     if (body == null) {
       this.body = null
     } else if (util.isStream(body)) {
+      const onEnd = () => {
+        util.destroy(body)
+      }
+
+      const onError = err => {
+        if (this.abort) {
+          this.abort(err)
+        } else {
+          this.errored = err
+        }
+      }
+
+      body
+        .on('end', onEnd)
+        .on('error', onError)
+
+      this.dispose = () => {
+        body
+          .off('end', onEnd)
+          .off('error', onError)
+      }
+
       this.body = body
     } else if (util.isBuffer(body)) {
       this.body = body.byteLength ? body : null
@@ -236,7 +260,12 @@ class Request {
     assert(!this.aborted)
     assert(!this.completed)
 
-    return this[kHandler].onConnect(abort)
+    if (this.errored) {
+      abort(this.errored)
+    } else {
+      this.abort = abort
+      return this[kHandler].onConnect(abort)
+    }
   }
 
   onHeaders (statusCode, headers, resume, statusText) {
@@ -267,6 +296,11 @@ class Request {
   onComplete (trailers) {
     assert(!this.aborted)
 
+    if (this.dispose) {
+      this.dispose()
+      this.dispose = null
+    }
+
     this.completed = true
     if (channels.trailers.hasSubscribers) {
       channels.trailers.publish({ request: this, trailers })
@@ -275,6 +309,11 @@ class Request {
   }
 
   onError (error) {
+    if (this.dispose) {
+      this.dispose()
+      this.dispose = null
+    }
+
     if (channels.error.hasSubscribers) {
       channels.error.publish({ request: this, error })
     }