commit

node_modules/undici/lib/api/api-request.js

@@ -3,10 +3,10 @@
 const Readable = require('./readable')
 const {
   InvalidArgumentError,
-  RequestAbortedError,
-  ResponseStatusCodeError
+  RequestAbortedError
 } = require('../core/errors')
 const util = require('../core/util')
+const { getResolveErrorBodyCallback } = require('./util')
 const { AsyncResource } = require('async_hooks')
 const { addSignal, removeSignal } = require('./abort-signal')
 
@@ -16,13 +16,17 @@ class RequestHandler extends AsyncResource {
       throw new InvalidArgumentError('invalid opts')
     }
 
-    const { signal, method, opaque, body, onInfo, responseHeaders, throwOnError } = opts
+    const { signal, method, opaque, body, onInfo, responseHeaders, throwOnError, highWaterMark } = opts
 
     try {
       if (typeof callback !== 'function') {
         throw new InvalidArgumentError('invalid callback')
       }
 
+      if (highWaterMark && (typeof highWaterMark !== 'number' || highWaterMark < 0)) {
+        throw new InvalidArgumentError('invalid highWaterMark')
+      }
+
       if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {
         throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget')
       }
@@ -53,6 +57,7 @@ class RequestHandler extends AsyncResource {
     this.context = null
     this.onInfo = onInfo || null
     this.throwOnError = throwOnError
+    this.highWaterMark = highWaterMark
 
     if (util.isStream(body)) {
       body.on('error', (err) => {
@@ -73,40 +78,39 @@ class RequestHandler extends AsyncResource {
   }
 
   onHeaders (statusCode, rawHeaders, resume, statusMessage) {
-    const { callback, opaque, abort, context } = this
+    const { callback, opaque, abort, context, responseHeaders, highWaterMark } = this
+
+    const headers = responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
 
     if (statusCode < 200) {
       if (this.onInfo) {
-        const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
         this.onInfo({ statusCode, headers })
       }
       return
     }
 
-    const parsedHeaders = util.parseHeaders(rawHeaders)
+    const parsedHeaders = responseHeaders === 'raw' ? util.parseHeaders(rawHeaders) : headers
     const contentType = parsedHeaders['content-type']
-    const body = new Readable(resume, abort, contentType)
+    const body = new Readable({ resume, abort, contentType, highWaterMark })
 
     this.callback = null
     this.res = body
-    const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
 
     if (callback !== null) {
       if (this.throwOnError && statusCode >= 400) {
         this.runInAsyncScope(getResolveErrorBodyCallback, null,
           { callback, body, contentType, statusCode, statusMessage, headers }
         )
-        return
+      } else {
+        this.runInAsyncScope(callback, null, null, {
+          statusCode,
+          headers,
+          trailers: this.trailers,
+          opaque,
+          body,
+          context
+        })
       }
-
-      this.runInAsyncScope(callback, null, null, {
-        statusCode,
-        headers,
-        trailers: this.trailers,
-        opaque,
-        body,
-        context
-      })
     }
   }
 
@@ -153,33 +157,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) => {

node_modules/undici/lib/api/api-stream.js

@@ -1,12 +1,13 @@
 'use strict'
 
-const { finished } = require('stream')
+const { finished, PassThrough } = require('stream')
 const {
   InvalidArgumentError,
   InvalidReturnValueError,
   RequestAbortedError
 } = require('../core/errors')
 const util = require('../core/util')
+const { getResolveErrorBodyCallback } = require('./util')
 const { AsyncResource } = require('async_hooks')
 const { addSignal, removeSignal } = require('./abort-signal')
 
@@ -16,7 +17,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 +58,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,52 +78,67 @@ 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, responseHeaders } = this
+
+    const headers = responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
 
     if (statusCode < 200) {
       if (this.onInfo) {
-        const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
         this.onInfo({ statusCode, headers })
       }
       return
     }
 
     this.factory = null
-    const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
-    const res = this.runInAsyncScope(factory, null, {
-      statusCode,
-      headers,
-      opaque,
-      context
-    })
 
-    if (
-      !res ||
-      typeof res.write !== 'function' ||
-      typeof res.end !== 'function' ||
-      typeof res.on !== 'function'
-    ) {
-      throw new InvalidReturnValueError('expected Writable')
+    let res
+
+    if (this.throwOnError && statusCode >= 400) {
+      const parsedHeaders = responseHeaders === 'raw' ? util.parseHeaders(rawHeaders) : headers
+      const contentType = parsedHeaders['content-type']
+      res = new PassThrough()
+
+      this.callback = null
+      this.runInAsyncScope(getResolveErrorBodyCallback, null,
+        { callback, body: res, contentType, statusCode, statusMessage, headers }
+      )
+    } else {
+      res = this.runInAsyncScope(factory, null, {
+        statusCode,
+        headers,
+        opaque,
+        context
+      })
+
+      if (
+        !res ||
+        typeof res.write !== 'function' ||
+        typeof res.end !== 'function' ||
+        typeof res.on !== 'function'
+      ) {
+        throw new InvalidReturnValueError('expected Writable')
+      }
+
+      // TODO: Avoid finished. It registers an unnecessary amount of listeners.
+      finished(res, { readable: false }, (err) => {
+        const { callback, res, opaque, trailers, abort } = this
+
+        this.res = null
+        if (err || !res.readable) {
+          util.destroy(res, err)
+        }
+
+        this.callback = null
+        this.runInAsyncScope(callback, null, err || null, { opaque, trailers })
+
+        if (err) {
+          abort()
+        }
+      })
     }
 
     res.on('drain', resume)
-    // TODO: Avoid finished. It registers an unnecessary amount of listeners.
-    finished(res, { readable: false }, (err) => {
-      const { callback, res, opaque, trailers, abort } = this
-
-      this.res = null
-      if (err || !res.readable) {
-        util.destroy(res, err)
-      }
-
-      this.callback = null
-      this.runInAsyncScope(callback, null, err || null, { opaque, trailers })
-
-      if (err) {
-        abort()
-      }
-    })
 
     this.res = res
 

node_modules/undici/lib/api/readable.js

@@ -4,7 +4,7 @@
 
 const assert = require('assert')
 const { Readable } = require('stream')
-const { RequestAbortedError, NotSupportedError } = require('../core/errors')
+const { RequestAbortedError, NotSupportedError, InvalidArgumentError } = require('../core/errors')
 const util = require('../core/util')
 const { ReadableStreamFrom, toUSVString } = require('../core/util')
 
@@ -17,11 +17,16 @@ const kAbort = Symbol('abort')
 const kContentType = Symbol('kContentType')
 
 module.exports = class BodyReadable extends Readable {
-  constructor (resume, abort, contentType = '') {
+  constructor ({
+    resume,
+    abort,
+    contentType = '',
+    highWaterMark = 64 * 1024 // Same as nodejs fs streams.
+  }) {
     super({
       autoDestroy: true,
       read: resume,
-      highWaterMark: 64 * 1024 // Same as nodejs fs streams.
+      highWaterMark
     })
 
     this._readableState.dataEmitted = false
@@ -146,15 +151,31 @@ module.exports = class BodyReadable extends Readable {
 
   async dump (opts) {
     let limit = opts && Number.isFinite(opts.limit) ? opts.limit : 262144
+    const signal = opts && opts.signal
+    const abortFn = () => {
+      this.destroy()
+    }
+    if (signal) {
+      if (typeof signal !== 'object' || !('aborted' in signal)) {
+        throw new InvalidArgumentError('signal must be an AbortSignal')
+      }
+      util.throwIfAborted(signal)
+      signal.addEventListener('abort', abortFn, { once: true })
+    }
     try {
       for await (const chunk of this) {
+        util.throwIfAborted(signal)
         limit -= Buffer.byteLength(chunk)
         if (limit < 0) {
           return
         }
       }
     } catch {
-      // Do nothing...
+      util.throwIfAborted(signal)
+    } finally {
+      if (signal) {
+        signal.removeEventListener('abort', abortFn)
+      }
     }
   }
 }

node_modules/undici/lib/api/util.js

@@ -0,0 +1,46 @@
+const assert = require('assert')
+const {
+  ResponseStatusCodeError
+} = require('../core/errors')
+const { toUSVString } = require('../core/util')
+
+async function getResolveErrorBodyCallback ({ callback, body, contentType, statusCode, statusMessage, headers }) {
+  assert(body)
+
+  let chunks = []
+  let limit = 0
+
+  for await (const chunk of body) {
+    chunks.push(chunk)
+    limit += chunk.length
+    if (limit > 128 * 1024) {
+      chunks = null
+      break
+    }
+  }
+
+  if (statusCode === 204 || !contentType || !chunks) {
+    process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers))
+    return
+  }
+
+  try {
+    if (contentType.startsWith('application/json')) {
+      const payload = JSON.parse(toUSVString(Buffer.concat(chunks)))
+      process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers, payload))
+      return
+    }
+
+    if (contentType.startsWith('text/')) {
+      const payload = toUSVString(Buffer.concat(chunks))
+      process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers, payload))
+      return
+    }
+  } catch (err) {
+    // Process in a fallback if error
+  }
+
+  process.nextTick(callback, new ResponseStatusCodeError(`Response status code ${statusCode}${statusMessage ? `: ${statusMessage}` : ''}`, statusCode, headers))
+}
+
+module.exports = { getResolveErrorBodyCallback }