fix: undici stream throwOnError

lib/api/api-request.js

@@ -3,8 +3,7 @@
 const Readable = require('./readable')
 const {
   InvalidArgumentError,
-  RequestAbortedError,
-  ResponseStatusCodeError
+  RequestAbortedError
 } = require('../core/errors')
 const util = require('../core/util')
 const { AsyncResource } = require('async_hooks')
@@ -93,7 +92,7 @@ class RequestHandler extends AsyncResource {
 
     if (callback !== null) {
       if (this.throwOnError && statusCode >= 400) {
-        this.runInAsyncScope(getResolveErrorBodyCallback, null,
+        this.runInAsyncScope(util.getResolveErrorBodyCallback, null,
           { callback, body, contentType, statusCode, statusMessage, headers }
         )
         return
@@ -153,33 +152,6 @@ 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) => {

lib/api/api-stream.js

@@ -1,6 +1,6 @@
 'use strict'
 
-const { finished } = require('stream')
+const { finished, PassThrough } = require('stream')
 const {
   InvalidArgumentError,
   InvalidReturnValueError,
@@ -16,7 +16,7 @@ class StreamHandler extends AsyncResource {
       throw new InvalidArgumentError('invalid opts')
     }
 
-    const { signal, method, opaque, body, onInfo, responseHeaders } = opts
+    const { signal, method, opaque, body, onInfo, responseHeaders, throwOnError } = opts
 
     try {
       if (typeof callback !== 'function') {
@@ -57,6 +57,7 @@ class StreamHandler extends AsyncResource {
     this.trailers = null
     this.body = body
     this.onInfo = onInfo || null
+    this.throwOnError = throwOnError || false
 
     if (util.isStream(body)) {
       body.on('error', (err) => {
@@ -76,8 +77,8 @@ class StreamHandler extends AsyncResource {
     this.context = context
   }
 
-  onHeaders (statusCode, rawHeaders, resume) {
-    const { factory, opaque, context } = this
+  onHeaders (statusCode, rawHeaders, resume, statusMessage) {
+    const { factory, opaque, context, callback } = this
 
     if (statusCode < 200) {
       if (this.onInfo) {
@@ -96,6 +97,24 @@ class StreamHandler extends AsyncResource {
       context
     })
 
+    if (this.throwOnError && statusCode >= 400) {
+      const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
+      const chunks = []
+      const pass = new PassThrough()
+      pass
+        .on('data', (chunk) => chunks.push(chunk))
+        .on('end', () => {
+          const payload = Buffer.concat(chunks).toString('utf8')
+          this.runInAsyncScope(
+            util.getResolveErrorBodyCallback,
+            null,
+            { callback, statusCode, statusMessage, headers, payload }
+          )
+        })
+      this.res = pass
+      return
+    }
+
     if (
       !res ||
       typeof res.write !== 'function' ||

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 } = require('./errors')
+const { InvalidArgumentError, ResponseStatusCodeError } = require('./errors')
 const { Blob } = require('buffer')
 const nodeUtil = require('util')
 const { stringify } = require('querystring')
@@ -342,6 +342,38 @@ function getSocketInfo (socket) {
   }
 }
 
+async function getResolveErrorBodyCallback ({ callback, body, contentType, statusCode, statusMessage, headers, payload }) {
+  if (typeof payload === 'string' && statusCode >= 400) {
+    process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers, payload))
+    return
+  }
+
+  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))
+}
+
 let ReadableStream
 function ReadableStreamFrom (iterable) {
   if (!ReadableStream) {
@@ -426,6 +458,7 @@ module.exports = {
   getSocketInfo,
   isFormDataLike,
   buildURL,
+  getResolveErrorBodyCallback,
   nodeMajor,
   nodeMinor,
   nodeHasAutoSelectFamily: nodeMajor > 18 || (nodeMajor === 18 && nodeMinor >= 13)

test/async_hooks.js

@@ -157,7 +157,7 @@ test('async hooks client is destroyed', (t) => {
     const client = new Client(`http://localhost:${server.address().port}`)
     t.teardown(client.destroy.bind(client))
 
-    client.request({ path: '/', method: 'GET' }, (err, { body }) => {
+    client.request({ path: '/', method: 'GET', throwOnError: true }, (err, { body }) => {
       t.error(err)
       body.resume()
       body.on('error', (err) => {

test/client-stream.js

@@ -785,4 +785,38 @@ test('stream legacy needDrain', (t) => {
       t.pass()
     })
   })
+
+  test('steam throwOnError', (t) => {
+    t.plan(2)
+
+    const errStatusCode = 500
+    const errMessage = 'Internal Server Error'
+
+    const server = createServer((req, res) => {
+      res.writeHead(errStatusCode, { 'Content-Type': 'text/plain' })
+      res.end(errMessage)
+    })
+    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))
+
+      client.stream({
+        path: '/',
+        method: 'GET',
+        throwOnError: true,
+        opaque: new PassThrough()
+      }, ({ opaque: pt }) => {
+        pt.on('data', () => {
+          t.not.pass()
+        })
+        return pt
+      }, (e) => {
+        t.equal(e.status, errStatusCode)
+        t.equal(e.body, errMessage)
+        t.end()
+      })
+    })
+  })
 })