fix(api-stream): Handle body correctly before throwing error

nodejs · Nov 12, 2022 · 8f7a6d7 · 8f7a6d7
1 parent 1307b84
commit 8f7a6d7
Show file tree

Hide file tree

Showing 4 changed files with 54 additions and 52 deletions.
diff --git a/lib/api/api-request.js b/lib/api/api-request.js
@@ -3,7 +3,8 @@
 const Readable = require('./readable')
 const {
   InvalidArgumentError,
-  RequestAbortedError
+  RequestAbortedError,
+  ResponseStatusCodeError
 } = require('../core/errors')
 const util = require('../core/util')
 const { AsyncResource } = require('async_hooks')
@@ -92,7 +93,7 @@ class RequestHandler extends AsyncResource {
 
     if (callback !== null) {
       if (this.throwOnError && statusCode >= 400) {
-        this.runInAsyncScope(util.getResolveErrorBodyCallback, null,
+        this.runInAsyncScope(getResolveErrorBodyCallback, null,
           { callback, body, contentType, statusCode, statusMessage, headers }
         )
         return
@@ -152,6 +153,33 @@ class RequestHandler extends AsyncResource {
   }
 }
 
+async function getResolveErrorBodyCallback ({ callback, body, contentType, statusCode, statusMessage, headers }) {
+  if (statusCode === 204 || !contentType) {
+    body.dump()
+    process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers))
+    return
+  }
+
+  try {
+    if (contentType.startsWith('application/json')) {
+      const payload = await body.json()
+      process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers, payload))
+      return
+    }
+
+    if (contentType.startsWith('text/')) {
+      const payload = await body.text()
+      process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers, payload))
+      return
+    }
+  } catch (err) {
+    // Process in a fallback if error
+  }
+
+  body.dump()
+  process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers))
+}
+
 function request (opts, callback) {
   if (callback === undefined) {
     return new Promise((resolve, reject) => {

diff --git a/lib/api/api-stream.js b/lib/api/api-stream.js
@@ -4,12 +4,11 @@ const { finished } = require('stream')
 const {
   InvalidArgumentError,
   InvalidReturnValueError,
-  RequestAbortedError
+  RequestAbortedError, ResponseStatusCodeError
 } = require('../core/errors')
 const util = require('../core/util')
 const { AsyncResource } = require('async_hooks')
 const { addSignal, removeSignal } = require('./abort-signal')
-const Readable = require('./readable')
 
 class StreamHandler extends AsyncResource {
   constructor (opts, factory, callback) {
@@ -79,7 +78,7 @@ class StreamHandler extends AsyncResource {
   }
 
   onHeaders (statusCode, rawHeaders, resume, statusMessage) {
-    const { factory, opaque, context, abort, callback } = this
+    const { factory, opaque, context } = this
 
     if (statusCode < 200) {
       if (this.onInfo) {
@@ -89,22 +88,8 @@ class StreamHandler extends AsyncResource {
       return
     }
 
-    const parsedHeaders = util.parseHeaders(rawHeaders)
-    const contentType = parsedHeaders['content-type']
-
     this.factory = null
     const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
-
-    if (this.throwOnError && statusCode >= 400) {
-      const body = new Readable(resume, abort, contentType)
-      this.res = body
-
-      this.runInAsyncScope(util.getResolveErrorBodyCallback, null,
-        { callback, body, contentType, statusCode, statusMessage, headers }
-      )
-      return
-    }
-
     const res = this.runInAsyncScope(factory, null, {
       statusCode,
       headers,
@@ -132,6 +117,12 @@ class StreamHandler extends AsyncResource {
       }
 
       this.callback = null
+
+      if (this.throwOnError && statusCode >= 400) {
+        this.runInAsyncScope(callback, null, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers))
+        return
+      }
+
       this.runInAsyncScope(callback, null, err || null, { opaque, trailers })
 
       if (err) {

diff --git a/lib/core/util.js b/lib/core/util.js
@@ -5,7 +5,7 @@ const { kDestroyed, kBodyUsed } = require('./symbols')
 const { IncomingMessage } = require('http')
 const stream = require('stream')
 const net = require('net')
-const { InvalidArgumentError, ResponseStatusCodeError } = require('./errors')
+const { InvalidArgumentError } = require('./errors')
 const { Blob } = require('buffer')
 const nodeUtil = require('util')
 const { stringify } = require('querystring')
@@ -361,33 +361,6 @@ function isFormDataLike (chunk) {
 const kEnumerableProperty = Object.create(null)
 kEnumerableProperty.enumerable = true
 
-async function getResolveErrorBodyCallback ({ callback, body, contentType, statusCode, statusMessage, headers }) {
-  if (statusCode === 204 || !contentType) {
-    body.dump()
-    process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers))
-    return
-  }
-
-  try {
-    if (contentType.startsWith('application/json')) {
-      const payload = await body.json()
-      process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers, payload))
-      return
-    }
-
-    if (contentType.startsWith('text/')) {
-      const payload = await body.text()
-      process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers, payload))
-      return
-    }
-  } catch (err) {
-    // Process in a fallback if error
-  }
-
-  body.dump()
-  process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers))
-}
-
 module.exports = {
   kEnumerableProperty,
   nop,
@@ -415,6 +388,5 @@ module.exports = {
   validateHandler,
   getSocketInfo,
   isFormDataLike,
-  buildURL,
-  getResolveErrorBodyCallback
+  buildURL
 }
diff --git a/test/client-stream.js b/test/client-stream.js
@@ -790,27 +790,38 @@ test('stream legacy needDrain', (t) => {
 test('steam throw if statusCode >= 400', (t) => {
   t.plan(1)
 
+  const expectedBodyContent = 'expected_body_content'
+
   const server = createServer((req, res) => {
     res.writeHead(400, { 'Content-Type': 'text/plain' })
-    res.end()
+    res.end(expectedBodyContent)
   })
   t.teardown(server.close.bind(server))
 
   server.listen(0, async () => {
     const client = new Client(`http://localhost:${server.address().port}`)
     t.teardown(client.close.bind(client))
 
+    const passThrough = new PassThrough()
+    const chunks = []
+
+    passThrough.on('data', (chunk) => {
+      chunks.push(chunk)
+    })
+
     try {
       await client.stream({
         path: '/',
         method: 'GET',
         throwOnError: true,
-        opaque: new PassThrough()
+        opaque: passThrough
       }, ({ opaque }) => opaque)
 
       t.fail('No Error')
     } catch (e) {
-      t.pass('Error')
+      const actualBodyContent = Buffer.concat(chunks).toString()
+
+      t.equal(actualBodyContent, expectedBodyContent)
     }
   })
 })