fix: handle request body as late as possible

Author: ronag

lib/client.js

@@ -105,6 +105,8 @@ const {
 // Experimental
 let h2ExperimentalWarned = false
 
+let extractBody
+
 const FastBuffer = Buffer[Symbol.species]
 
 const kClosedResolve = Symbol('kClosedResolve')
@@ -1477,7 +1479,9 @@ function write (client, request) {
     return
   }
 
-  const { body, method, path, host, upgrade, headers, blocking, reset } = request
+  const { method, path, host, upgrade, blocking, reset } = request
+
+  let { body, headers, contentLength } = request
 
   // https://tools.ietf.org/html/rfc7231#section-4.3.1
   // https://tools.ietf.org/html/rfc7231#section-4.3.2
@@ -1494,14 +1498,42 @@ function write (client, request) {
     method === 'PATCH'
   )
 
+  if (util.isFormDataLike(body)) {
+    if (!extractBody) {
+      extractBody = require('./fetch/body.js').extractBody
+    }
+
+    const [bodyStream, contentType] = extractBody(body)
+    if (request.contentType == null) {
+      headers += `content-type: ${contentType}\r\n`
+    }
+    body = bodyStream.stream
+    contentLength = bodyStream.length
+  } else if (util.isBlobLike(body) && request.contentType == null && body.type) {
+    headers += `content-type: ${body.type}\r\n`
+  }
+
+  // TODO (fix): Error if stream is disturbed?
+
+  if (util.isStream(body) && (body.readableEnded || body.errored)) {
+    let err = body.errored
+    if (!err) {
+      err = new RequestAbortedError()
+      util.destroy(body, err)
+    }
+    body.on('error', () => {})
+    errorRequest(client, request, err)
+    return false
+  }
+
   if (body && typeof body.read === 'function') {
     // Try to read EOF in order to get length.
     body.read(0)
   }
 
   const bodyLength = util.bodyLength(body)
 
-  let contentLength = bodyLength
+  contentLength = bodyLength ?? contentLength
 
   if (contentLength === null) {
     contentLength = request.contentLength
@@ -1544,6 +1576,7 @@ function write (client, request) {
   }
 
   if (request.aborted) {
+    util.destroy(body)
     return false
   }
 
@@ -2050,6 +2083,10 @@ function writeStream ({ h2stream, body, client, request, socket, contentLength,
   socket
     .on('drain', onDrain)
     .on('error', onFinished)
+
+  if (body.closeEmitted ?? body.closed) {
+    queueMicrotask(onClose)
+  }
 }
 
 async function writeBlob ({ h2stream, body, client, request, socket, contentLength, header, expectsPayload }) {

lib/core/request.js

@@ -26,8 +26,6 @@ const invalidPathRegex = /[^\u0021-\u00ff]/
 
 const kHandler = Symbol('handler')
 
-let extractBody
-
 class Request {
   constructor (origin, {
     path,
@@ -173,23 +171,6 @@ class Request {
       throw new InvalidArgumentError('headers must be an object or an array')
     }
 
-    if (util.isFormDataLike(this.body)) {
-      if (!extractBody) {
-        extractBody = require('../fetch/body.js').extractBody
-      }
-
-      const [bodyStream, contentType] = extractBody(body)
-      if (this.contentType == null) {
-        this.contentType = contentType
-        this.headers += `content-type: ${contentType}\r\n`
-      }
-      this.body = bodyStream.stream
-      this.contentLength = bodyStream.length
-    } else if (util.isBlobLike(body) && this.contentType == null && body.type) {
-      this.contentType = body.type
-      this.headers += `content-type: ${body.type}\r\n`
-    }
-
     util.validateHandler(handler, method, upgrade)
 
     this.servername = util.getServerName(this.host)

lib/core/util.js

@@ -182,8 +182,8 @@ function bodyLength (body) {
   return null
 }
 
-function isDestroyed (stream) {
-  return !stream || !!(stream.destroyed || stream[kDestroyed])
+function isDestroyed (body) {
+  return body && !!(body.destroyed || body[kDestroyed] || (stream.isDestroyed?.(body)))
 }
 
 function isReadableAborted (stream) {

test/client.js

@@ -1055,6 +1055,7 @@ test('basic POST with empty stream', (t) => {
       method: 'POST',
       body
     }, (err, { statusCode, headers, body }) => {
+      console.error(err)
       t.error(err)
       body
         .on('data', () => {