fix(api-stream): Add throwOnError support in undici.stream

lib/api/api-stream.js

@@ -4,7 +4,7 @@ const { finished } = require('stream')
 const {
  InvalidArgumentError,
  InvalidReturnValueError,
- RequestAbortedError
+ RequestAbortedError, ResponseStatusCodeError
 } = require('../core/errors')
 const util = require('../core/util')
 const { AsyncResource } = require('async_hooks')
@@ -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
 
  if (util.isStream(body)) {
  body.on('error', (err) => {
@@ -76,7 +77,7 @@ class StreamHandler extends AsyncResource {
  this.context = context
  }
 
- onHeaders (statusCode, rawHeaders, resume) {
+ onHeaders (statusCode, rawHeaders, resume, statusMessage) {
  const { factory, opaque, context } = this
 
  if (statusCode < 200) {
@@ -116,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) {

lib/core/util.js

@@ -282,9 +282,9 @@ function isDisturbed (body) {
  stream.isDisturbed
  ? stream.isDisturbed(body) || body[kBodyUsed] // TODO (fix): Why is body[kBodyUsed] needed?
  : body[kBodyUsed] ||
-  body.readableDidRead ||
-  (body._readableState && body._readableState.dataEmitted) ||
-  isReadableAborted(body)
+ body.readableDidRead ||
+ (body._readableState && body._readableState.dataEmitted) ||
+ isReadableAborted(body)
  ))
 }
 

test/client-stream.js

@@ -786,3 +786,42 @@ 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(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: passThrough
+ }, ({ opaque }) => opaque)
+
+ t.fail('No Error')
+ } catch (e) {
+ const actualBodyContent = Buffer.concat(chunks).toString()
+
+ t.equal(actualBodyContent, expectedBodyContent)
+ }
+ })
+})