Adding upstream version 5.28.2+dfsg1+~cs23.11.12.3.upstream/5.28.2+dfsg1+_cs23.11.12.3upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

9 files changed, 1287 insertions, 0 deletions

diff --git a/lib/api/abort-signal.js b/lib/api/abort-signal.js
new file mode 100644
index 0000000..2985c1e
--- /dev/null
+++ b/lib/api/abort-signal.js
@@ -0,0 +1,54 @@
+const { addAbortListener } = require('../core/util')
+const { RequestAbortedError } = require('../core/errors')
+
+const kListener = Symbol('kListener')
+const kSignal = Symbol('kSignal')
+
+function abort (self) {
+  if (self.abort) {
+    self.abort()
+  } else {
+    self.onError(new RequestAbortedError())
+  }
+}
+
+function addSignal (self, signal) {
+  self[kSignal] = null
+  self[kListener] = null
+
+  if (!signal) {
+    return
+  }
+
+  if (signal.aborted) {
+    abort(self)
+    return
+  }
+
+  self[kSignal] = signal
+  self[kListener] = () => {
+    abort(self)
+  }
+
+  addAbortListener(self[kSignal], self[kListener])
+}
+
+function removeSignal (self) {
+  if (!self[kSignal]) {
+    return
+  }
+
+  if ('removeEventListener' in self[kSignal]) {
+    self[kSignal].removeEventListener('abort', self[kListener])
+  } else {
+    self[kSignal].removeListener('abort', self[kListener])
+  }
+
+  self[kSignal] = null
+  self[kListener] = null
+}
+
+module.exports = {
+  addSignal,
+  removeSignal
+}
diff --git a/lib/api/api-connect.js b/lib/api/api-connect.js
new file mode 100644
index 0000000..fd2b6ad
--- /dev/null
+++ b/lib/api/api-connect.js
@@ -0,0 +1,104 @@
+'use strict'
+
+const { AsyncResource } = require('async_hooks')
+const { InvalidArgumentError, RequestAbortedError, SocketError } = require('../core/errors')
+const util = require('../core/util')
+const { addSignal, removeSignal } = require('./abort-signal')
+
+class ConnectHandler extends AsyncResource {
+  constructor (opts, callback) {
+    if (!opts || typeof opts !== 'object') {
+      throw new InvalidArgumentError('invalid opts')
+    }
+
+    if (typeof callback !== 'function') {
+      throw new InvalidArgumentError('invalid callback')
+    }
+
+    const { signal, opaque, responseHeaders } = opts
+
+    if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {
+      throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget')
+    }
+
+    super('UNDICI_CONNECT')
+
+    this.opaque = opaque || null
+    this.responseHeaders = responseHeaders || null
+    this.callback = callback
+    this.abort = null
+
+    addSignal(this, signal)
+  }
+
+  onConnect (abort, context) {
+    if (!this.callback) {
+      throw new RequestAbortedError()
+    }
+
+    this.abort = abort
+    this.context = context
+  }
+
+  onHeaders () {
+    throw new SocketError('bad connect', null)
+  }
+
+  onUpgrade (statusCode, rawHeaders, socket) {
+    const { callback, opaque, context } = this
+
+    removeSignal(this)
+
+    this.callback = null
+
+    let headers = rawHeaders
+    // Indicates is an HTTP2Session
+    if (headers != null) {
+      headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
+    }
+
+    this.runInAsyncScope(callback, null, null, {
+      statusCode,
+      headers,
+      socket,
+      opaque,
+      context
+    })
+  }
+
+  onError (err) {
+    const { callback, opaque } = this
+
+    removeSignal(this)
+
+    if (callback) {
+      this.callback = null
+      queueMicrotask(() => {
+        this.runInAsyncScope(callback, null, err, { opaque })
+      })
+    }
+  }
+}
+
+function connect (opts, callback) {
+  if (callback === undefined) {
+    return new Promise((resolve, reject) => {
+      connect.call(this, opts, (err, data) => {
+        return err ? reject(err) : resolve(data)
+      })
+    })
+  }
+
+  try {
+    const connectHandler = new ConnectHandler(opts, callback)
+    this.dispatch({ ...opts, method: 'CONNECT' }, connectHandler)
+  } catch (err) {
+    if (typeof callback !== 'function') {
+      throw err
+    }
+    const opaque = opts && opts.opaque
+    queueMicrotask(() => callback(err, { opaque }))
+  }
+}
+
+module.exports = connect
diff --git a/lib/api/api-pipeline.js b/lib/api/api-pipeline.js
new file mode 100644
index 0000000..af4a180
--- /dev/null
+++ b/lib/api/api-pipeline.js
@@ -0,0 +1,249 @@
+'use strict'
+
+const {
+  Readable,
+  Duplex,
+  PassThrough
+} = require('stream')
+const {
+  InvalidArgumentError,
+  InvalidReturnValueError,
+  RequestAbortedError
+} = require('../core/errors')
+const util = require('../core/util')
+const { AsyncResource } = require('async_hooks')
+const { addSignal, removeSignal } = require('./abort-signal')
+const assert = require('assert')
+
+const kResume = Symbol('resume')
+
+class PipelineRequest extends Readable {
+  constructor () {
+    super({ autoDestroy: true })
+
+    this[kResume] = null
+  }
+
+  _read () {
+    const { [kResume]: resume } = this
+
+    if (resume) {
+      this[kResume] = null
+      resume()
+    }
+  }
+
+  _destroy (err, callback) {
+    this._read()
+
+    callback(err)
+  }
+}
+
+class PipelineResponse extends Readable {
+  constructor (resume) {
+    super({ autoDestroy: true })
+    this[kResume] = resume
+  }
+
+  _read () {
+    this[kResume]()
+  }
+
+  _destroy (err, callback) {
+    if (!err && !this._readableState.endEmitted) {
+      err = new RequestAbortedError()
+    }
+
+    callback(err)
+  }
+}
+
+class PipelineHandler extends AsyncResource {
+  constructor (opts, handler) {
+    if (!opts || typeof opts !== 'object') {
+      throw new InvalidArgumentError('invalid opts')
+    }
+
+    if (typeof handler !== 'function') {
+      throw new InvalidArgumentError('invalid handler')
+    }
+
+    const { signal, method, opaque, onInfo, responseHeaders } = opts
+
+    if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {
+      throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget')
+    }
+
+    if (method === 'CONNECT') {
+      throw new InvalidArgumentError('invalid method')
+    }
+
+    if (onInfo && typeof onInfo !== 'function') {
+      throw new InvalidArgumentError('invalid onInfo callback')
+    }
+
+    super('UNDICI_PIPELINE')
+
+    this.opaque = opaque || null
+    this.responseHeaders = responseHeaders || null
+    this.handler = handler
+    this.abort = null
+    this.context = null
+    this.onInfo = onInfo || null
+
+    this.req = new PipelineRequest().on('error', util.nop)
+
+    this.ret = new Duplex({
+      readableObjectMode: opts.objectMode,
+      autoDestroy: true,
+      read: () => {
+        const { body } = this
+
+        if (body && body.resume) {
+          body.resume()
+        }
+      },
+      write: (chunk, encoding, callback) => {
+        const { req } = this
+
+        if (req.push(chunk, encoding) || req._readableState.destroyed) {
+          callback()
+        } else {
+          req[kResume] = callback
+        }
+      },
+      destroy: (err, callback) => {
+        const { body, req, res, ret, abort } = this
+
+        if (!err && !ret._readableState.endEmitted) {
+          err = new RequestAbortedError()
+        }
+
+        if (abort && err) {
+          abort()
+        }
+
+        util.destroy(body, err)
+        util.destroy(req, err)
+        util.destroy(res, err)
+
+        removeSignal(this)
+
+        callback(err)
+      }
+    }).on('prefinish', () => {
+      const { req } = this
+
+      // Node < 15 does not call _final in same tick.
+      req.push(null)
+    })
+
+    this.res = null
+
+    addSignal(this, signal)
+  }
+
+  onConnect (abort, context) {
+    const { ret, res } = this
+
+    assert(!res, 'pipeline cannot be retried')
+
+    if (ret.destroyed) {
+      throw new RequestAbortedError()
+    }
+
+    this.abort = abort
+    this.context = context
+  }
+
+  onHeaders (statusCode, rawHeaders, resume) {
+    const { opaque, handler, context } = this
+
+    if (statusCode < 200) {
+      if (this.onInfo) {
+        const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
+        this.onInfo({ statusCode, headers })
+      }
+      return
+    }
+
+    this.res = new PipelineResponse(resume)
+
+    let body
+    try {
+      this.handler = null
+      const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
+      body = this.runInAsyncScope(handler, null, {
+        statusCode,
+        headers,
+        opaque,
+        body: this.res,
+        context
+      })
+    } catch (err) {
+      this.res.on('error', util.nop)
+      throw err
+    }
+
+    if (!body || typeof body.on !== 'function') {
+      throw new InvalidReturnValueError('expected Readable')
+    }
+
+    body
+      .on('data', (chunk) => {
+        const { ret, body } = this
+
+        if (!ret.push(chunk) && body.pause) {
+          body.pause()
+        }
+      })
+      .on('error', (err) => {
+        const { ret } = this
+
+        util.destroy(ret, err)
+      })
+      .on('end', () => {
+        const { ret } = this
+
+        ret.push(null)
+      })
+      .on('close', () => {
+        const { ret } = this
+
+        if (!ret._readableState.ended) {
+          util.destroy(ret, new RequestAbortedError())
+        }
+      })
+
+    this.body = body
+  }
+
+  onData (chunk) {
+    const { res } = this
+    return res.push(chunk)
+  }
+
+  onComplete (trailers) {
+    const { res } = this
+    res.push(null)
+  }
+
+  onError (err) {
+    const { ret } = this
+    this.handler = null
+    util.destroy(ret, err)
+  }
+}
+
+function pipeline (opts, handler) {
+  try {
+    const pipelineHandler = new PipelineHandler(opts, handler)
+    this.dispatch({ ...opts, body: pipelineHandler.req }, pipelineHandler)
+    return pipelineHandler.ret
+  } catch (err) {
+    return new PassThrough().destroy(err)
+  }
+}
+
+module.exports = pipeline
diff --git a/lib/api/api-request.js b/lib/api/api-request.js
new file mode 100644
index 0000000..d4281ce
--- /dev/null
+++ b/lib/api/api-request.js
@@ -0,0 +1,180 @@
+'use strict'
+
+const Readable = require('./readable')
+const {
+  InvalidArgumentError,
+  RequestAbortedError
+} = require('../core/errors')
+const util = require('../core/util')
+const { getResolveErrorBodyCallback } = require('./util')
+const { AsyncResource } = require('async_hooks')
+const { addSignal, removeSignal } = require('./abort-signal')
+
+class RequestHandler extends AsyncResource {
+  constructor (opts, callback) {
+    if (!opts || typeof opts !== 'object') {
+      throw new InvalidArgumentError('invalid 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')
+      }
+
+      if (method === 'CONNECT') {
+        throw new InvalidArgumentError('invalid method')
+      }
+
+      if (onInfo && typeof onInfo !== 'function') {
+        throw new InvalidArgumentError('invalid onInfo callback')
+      }
+
+      super('UNDICI_REQUEST')
+    } catch (err) {
+      if (util.isStream(body)) {
+        util.destroy(body.on('error', util.nop), err)
+      }
+      throw err
+    }
+
+    this.responseHeaders = responseHeaders || null
+    this.opaque = opaque || null
+    this.callback = callback
+    this.res = null
+    this.abort = null
+    this.body = body
+    this.trailers = {}
+    this.context = null
+    this.onInfo = onInfo || null
+    this.throwOnError = throwOnError
+    this.highWaterMark = highWaterMark
+
+    if (util.isStream(body)) {
+      body.on('error', (err) => {
+        this.onError(err)
+      })
+    }
+
+    addSignal(this, signal)
+  }
+
+  onConnect (abort, context) {
+    if (!this.callback) {
+      throw new RequestAbortedError()
+    }
+
+    this.abort = abort
+    this.context = context
+  }
+
+  onHeaders (statusCode, rawHeaders, resume, statusMessage) {
+    const { callback, opaque, abort, context, responseHeaders, highWaterMark } = this
+
+    const headers = responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
+
+    if (statusCode < 200) {
+      if (this.onInfo) {
+        this.onInfo({ statusCode, headers })
+      }
+      return
+    }
+
+    const parsedHeaders = responseHeaders === 'raw' ? util.parseHeaders(rawHeaders) : headers
+    const contentType = parsedHeaders['content-type']
+    const body = new Readable({ resume, abort, contentType, highWaterMark })
+
+    this.callback = null
+    this.res = body
+    if (callback !== null) {
+      if (this.throwOnError && statusCode >= 400) {
+        this.runInAsyncScope(getResolveErrorBodyCallback, null,
+          { callback, body, contentType, statusCode, statusMessage, headers }
+        )
+      } else {
+        this.runInAsyncScope(callback, null, null, {
+          statusCode,
+          headers,
+          trailers: this.trailers,
+          opaque,
+          body,
+          context
+        })
+      }
+    }
+  }
+
+  onData (chunk) {
+    const { res } = this
+    return res.push(chunk)
+  }
+
+  onComplete (trailers) {
+    const { res } = this
+
+    removeSignal(this)
+
+    util.parseHeaders(trailers, this.trailers)
+
+    res.push(null)
+  }
+
+  onError (err) {
+    const { res, callback, body, opaque } = this
+
+    removeSignal(this)
+
+    if (callback) {
+      // TODO: Does this need queueMicrotask?
+      this.callback = null
+      queueMicrotask(() => {
+        this.runInAsyncScope(callback, null, err, { opaque })
+      })
+    }
+
+    if (res) {
+      this.res = null
+      // Ensure all queued handlers are invoked before destroying res.
+      queueMicrotask(() => {
+        util.destroy(res, err)
+      })
+    }
+
+    if (body) {
+      this.body = null
+      util.destroy(body, err)
+    }
+  }
+}
+
+function request (opts, callback) {
+  if (callback === undefined) {
+    return new Promise((resolve, reject) => {
+      request.call(this, opts, (err, data) => {
+        return err ? reject(err) : resolve(data)
+      })
+    })
+  }
+
+  try {
+    this.dispatch(opts, new RequestHandler(opts, callback))
+  } catch (err) {
+    if (typeof callback !== 'function') {
+      throw err
+    }
+    const opaque = opts && opts.opaque
+    queueMicrotask(() => callback(err, { opaque }))
+  }
+}
+
+module.exports = request
+module.exports.RequestHandler = RequestHandler
diff --git a/lib/api/api-stream.js b/lib/api/api-stream.js
new file mode 100644
index 0000000..c571a6f
--- /dev/null
+++ b/lib/api/api-stream.js
@@ -0,0 +1,220 @@
+'use strict'
+
+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')
+
+class StreamHandler extends AsyncResource {
+  constructor (opts, factory, callback) {
+    if (!opts || typeof opts !== 'object') {
+      throw new InvalidArgumentError('invalid opts')
+    }
+
+    const { signal, method, opaque, body, onInfo, responseHeaders, throwOnError } = opts
+
+    try {
+      if (typeof callback !== 'function') {
+        throw new InvalidArgumentError('invalid callback')
+      }
+
+      if (typeof factory !== 'function') {
+        throw new InvalidArgumentError('invalid factory')
+      }
+
+      if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {
+        throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget')
+      }
+
+      if (method === 'CONNECT') {
+        throw new InvalidArgumentError('invalid method')
+      }
+
+      if (onInfo && typeof onInfo !== 'function') {
+        throw new InvalidArgumentError('invalid onInfo callback')
+      }
+
+      super('UNDICI_STREAM')
+    } catch (err) {
+      if (util.isStream(body)) {
+        util.destroy(body.on('error', util.nop), err)
+      }
+      throw err
+    }
+
+    this.responseHeaders = responseHeaders || null
+    this.opaque = opaque || null
+    this.factory = factory
+    this.callback = callback
+    this.res = null
+    this.abort = null
+    this.context = null
+    this.trailers = null
+    this.body = body
+    this.onInfo = onInfo || null
+    this.throwOnError = throwOnError || false
+
+    if (util.isStream(body)) {
+      body.on('error', (err) => {
+        this.onError(err)
+      })
+    }
+
+    addSignal(this, signal)
+  }
+
+  onConnect (abort, context) {
+    if (!this.callback) {
+      throw new RequestAbortedError()
+    }
+
+    this.abort = abort
+    this.context = context
+  }
+
+  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) {
+        this.onInfo({ statusCode, headers })
+      }
+      return
+    }
+
+    this.factory = null
+
+    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 {
+      if (factory === null) {
+        return
+      }
+
+      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)
+
+    this.res = res
+
+    const needDrain = res.writableNeedDrain !== undefined
+      ? res.writableNeedDrain
+      : res._writableState && res._writableState.needDrain
+
+    return needDrain !== true
+  }
+
+  onData (chunk) {
+    const { res } = this
+
+    return res ? res.write(chunk) : true
+  }
+
+  onComplete (trailers) {
+    const { res } = this
+
+    removeSignal(this)
+
+    if (!res) {
+      return
+    }
+
+    this.trailers = util.parseHeaders(trailers)
+
+    res.end()
+  }
+
+  onError (err) {
+    const { res, callback, opaque, body } = this
+
+    removeSignal(this)
+
+    this.factory = null
+
+    if (res) {
+      this.res = null
+      util.destroy(res, err)
+    } else if (callback) {
+      this.callback = null
+      queueMicrotask(() => {
+        this.runInAsyncScope(callback, null, err, { opaque })
+      })
+    }
+
+    if (body) {
+      this.body = null
+      util.destroy(body, err)
+    }
+  }
+}
+
+function stream (opts, factory, callback) {
+  if (callback === undefined) {
+    return new Promise((resolve, reject) => {
+      stream.call(this, opts, factory, (err, data) => {
+        return err ? reject(err) : resolve(data)
+      })
+    })
+  }
+
+  try {
+    this.dispatch(opts, new StreamHandler(opts, factory, callback))
+  } catch (err) {
+    if (typeof callback !== 'function') {
+      throw err
+    }
+    const opaque = opts && opts.opaque
+    queueMicrotask(() => callback(err, { opaque }))
+  }
+}
+
+module.exports = stream
diff --git a/lib/api/api-upgrade.js b/lib/api/api-upgrade.js
new file mode 100644
index 0000000..ef783e8
--- /dev/null
+++ b/lib/api/api-upgrade.js
@@ -0,0 +1,105 @@
+'use strict'
+
+const { InvalidArgumentError, RequestAbortedError, SocketError } = require('../core/errors')
+const { AsyncResource } = require('async_hooks')
+const util = require('../core/util')
+const { addSignal, removeSignal } = require('./abort-signal')
+const assert = require('assert')
+
+class UpgradeHandler extends AsyncResource {
+  constructor (opts, callback) {
+    if (!opts || typeof opts !== 'object') {
+      throw new InvalidArgumentError('invalid opts')
+    }
+
+    if (typeof callback !== 'function') {
+      throw new InvalidArgumentError('invalid callback')
+    }
+
+    const { signal, opaque, responseHeaders } = opts
+
+    if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {
+      throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget')
+    }
+
+    super('UNDICI_UPGRADE')
+
+    this.responseHeaders = responseHeaders || null
+    this.opaque = opaque || null
+    this.callback = callback
+    this.abort = null
+    this.context = null
+
+    addSignal(this, signal)
+  }
+
+  onConnect (abort, context) {
+    if (!this.callback) {
+      throw new RequestAbortedError()
+    }
+
+    this.abort = abort
+    this.context = null
+  }
+
+  onHeaders () {
+    throw new SocketError('bad upgrade', null)
+  }
+
+  onUpgrade (statusCode, rawHeaders, socket) {
+    const { callback, opaque, context } = this
+
+    assert.strictEqual(statusCode, 101)
+
+    removeSignal(this)
+
+    this.callback = null
+    const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
+    this.runInAsyncScope(callback, null, null, {
+      headers,
+      socket,
+      opaque,
+      context
+    })
+  }
+
+  onError (err) {
+    const { callback, opaque } = this
+
+    removeSignal(this)
+
+    if (callback) {
+      this.callback = null
+      queueMicrotask(() => {
+        this.runInAsyncScope(callback, null, err, { opaque })
+      })
+    }
+  }
+}
+
+function upgrade (opts, callback) {
+  if (callback === undefined) {
+    return new Promise((resolve, reject) => {
+      upgrade.call(this, opts, (err, data) => {
+        return err ? reject(err) : resolve(data)
+      })
+    })
+  }
+
+  try {
+    const upgradeHandler = new UpgradeHandler(opts, callback)
+    this.dispatch({
+      ...opts,
+      method: opts.method || 'GET',
+      upgrade: opts.protocol || 'Websocket'
+    }, upgradeHandler)
+  } catch (err) {
+    if (typeof callback !== 'function') {
+      throw err
+    }
+    const opaque = opts && opts.opaque
+    queueMicrotask(() => callback(err, { opaque }))
+  }
+}
+
+module.exports = upgrade
diff --git a/lib/api/index.js b/lib/api/index.js
new file mode 100644
index 0000000..8983a5e
--- /dev/null
+++ b/lib/api/index.js
@@ -0,0 +1,7 @@
+'use strict'
+
+module.exports.request = require('./api-request')
+module.exports.stream = require('./api-stream')
+module.exports.pipeline = require('./api-pipeline')
+module.exports.upgrade = require('./api-upgrade')
+module.exports.connect = require('./api-connect')
diff --git a/lib/api/readable.js b/lib/api/readable.js
new file mode 100644
index 0000000..5269dfa
--- /dev/null
+++ b/lib/api/readable.js
@@ -0,0 +1,322 @@
+// Ported from https://github.com/nodejs/undici/pull/907
+
+'use strict'
+
+const assert = require('assert')
+const { Readable } = require('stream')
+const { RequestAbortedError, NotSupportedError, InvalidArgumentError } = require('../core/errors')
+const util = require('../core/util')
+const { ReadableStreamFrom, toUSVString } = require('../core/util')
+
+let Blob
+
+const kConsume = Symbol('kConsume')
+const kReading = Symbol('kReading')
+const kBody = Symbol('kBody')
+const kAbort = Symbol('abort')
+const kContentType = Symbol('kContentType')
+
+const noop = () => {}
+
+module.exports = class BodyReadable extends Readable {
+  constructor ({
+    resume,
+    abort,
+    contentType = '',
+    highWaterMark = 64 * 1024 // Same as nodejs fs streams.
+  }) {
+    super({
+      autoDestroy: true,
+      read: resume,
+      highWaterMark
+    })
+
+    this._readableState.dataEmitted = false
+
+    this[kAbort] = abort
+    this[kConsume] = null
+    this[kBody] = null
+    this[kContentType] = contentType
+
+    // Is stream being consumed through Readable API?
+    // This is an optimization so that we avoid checking
+    // for 'data' and 'readable' listeners in the hot path
+    // inside push().
+    this[kReading] = false
+  }
+
+  destroy (err) {
+    if (this.destroyed) {
+      // Node < 16
+      return this
+    }
+
+    if (!err && !this._readableState.endEmitted) {
+      err = new RequestAbortedError()
+    }
+
+    if (err) {
+      this[kAbort]()
+    }
+
+    return super.destroy(err)
+  }
+
+  emit (ev, ...args) {
+    if (ev === 'data') {
+      // Node < 16.7
+      this._readableState.dataEmitted = true
+    } else if (ev === 'error') {
+      // Node < 16
+      this._readableState.errorEmitted = true
+    }
+    return super.emit(ev, ...args)
+  }
+
+  on (ev, ...args) {
+    if (ev === 'data' || ev === 'readable') {
+      this[kReading] = true
+    }
+    return super.on(ev, ...args)
+  }
+
+  addListener (ev, ...args) {
+    return this.on(ev, ...args)
+  }
+
+  off (ev, ...args) {
+    const ret = super.off(ev, ...args)
+    if (ev === 'data' || ev === 'readable') {
+      this[kReading] = (
+        this.listenerCount('data') > 0 ||
+        this.listenerCount('readable') > 0
+      )
+    }
+    return ret
+  }
+
+  removeListener (ev, ...args) {
+    return this.off(ev, ...args)
+  }
+
+  push (chunk) {
+    if (this[kConsume] && chunk !== null && this.readableLength === 0) {
+      consumePush(this[kConsume], chunk)
+      return this[kReading] ? super.push(chunk) : true
+    }
+    return super.push(chunk)
+  }
+
+  // https://fetch.spec.whatwg.org/#dom-body-text
+  async text () {
+    return consume(this, 'text')
+  }
+
+  // https://fetch.spec.whatwg.org/#dom-body-json
+  async json () {
+    return consume(this, 'json')
+  }
+
+  // https://fetch.spec.whatwg.org/#dom-body-blob
+  async blob () {
+    return consume(this, 'blob')
+  }
+
+  // https://fetch.spec.whatwg.org/#dom-body-arraybuffer
+  async arrayBuffer () {
+    return consume(this, 'arrayBuffer')
+  }
+
+  // https://fetch.spec.whatwg.org/#dom-body-formdata
+  async formData () {
+    // TODO: Implement.
+    throw new NotSupportedError()
+  }
+
+  // https://fetch.spec.whatwg.org/#dom-body-bodyused
+  get bodyUsed () {
+    return util.isDisturbed(this)
+  }
+
+  // https://fetch.spec.whatwg.org/#dom-body-body
+  get body () {
+    if (!this[kBody]) {
+      this[kBody] = ReadableStreamFrom(this)
+      if (this[kConsume]) {
+        // TODO: Is this the best way to force a lock?
+        this[kBody].getReader() // Ensure stream is locked.
+        assert(this[kBody].locked)
+      }
+    }
+    return this[kBody]
+  }
+
+  dump (opts) {
+    let limit = opts && Number.isFinite(opts.limit) ? opts.limit : 262144
+    const signal = opts && opts.signal
+
+    if (signal) {
+      try {
+        if (typeof signal !== 'object' || !('aborted' in signal)) {
+          throw new InvalidArgumentError('signal must be an AbortSignal')
+        }
+        util.throwIfAborted(signal)
+      } catch (err) {
+        return Promise.reject(err)
+      }
+    }
+
+    if (this.closed) {
+      return Promise.resolve(null)
+    }
+
+    return new Promise((resolve, reject) => {
+      const signalListenerCleanup = signal
+        ? util.addAbortListener(signal, () => {
+          this.destroy()
+        })
+        : noop
+
+      this
+        .on('close', function () {
+          signalListenerCleanup()
+          if (signal && signal.aborted) {
+            reject(signal.reason || Object.assign(new Error('The operation was aborted'), { name: 'AbortError' }))
+          } else {
+            resolve(null)
+          }
+        })
+        .on('error', noop)
+        .on('data', function (chunk) {
+          limit -= chunk.length
+          if (limit <= 0) {
+            this.destroy()
+          }
+        })
+        .resume()
+    })
+  }
+}
+
+// https://streams.spec.whatwg.org/#readablestream-locked
+function isLocked (self) {
+  // Consume is an implicit lock.
+  return (self[kBody] && self[kBody].locked === true) || self[kConsume]
+}
+
+// https://fetch.spec.whatwg.org/#body-unusable
+function isUnusable (self) {
+  return util.isDisturbed(self) || isLocked(self)
+}
+
+async function consume (stream, type) {
+  if (isUnusable(stream)) {
+    throw new TypeError('unusable')
+  }
+
+  assert(!stream[kConsume])
+
+  return new Promise((resolve, reject) => {
+    stream[kConsume] = {
+      type,
+      stream,
+      resolve,
+      reject,
+      length: 0,
+      body: []
+    }
+
+    stream
+      .on('error', function (err) {
+        consumeFinish(this[kConsume], err)
+      })
+      .on('close', function () {
+        if (this[kConsume].body !== null) {
+          consumeFinish(this[kConsume], new RequestAbortedError())
+        }
+      })
+
+    process.nextTick(consumeStart, stream[kConsume])
+  })
+}
+
+function consumeStart (consume) {
+  if (consume.body === null) {
+    return
+  }
+
+  const { _readableState: state } = consume.stream
+
+  for (const chunk of state.buffer) {
+    consumePush(consume, chunk)
+  }
+
+  if (state.endEmitted) {
+    consumeEnd(this[kConsume])
+  } else {
+    consume.stream.on('end', function () {
+      consumeEnd(this[kConsume])
+    })
+  }
+
+  consume.stream.resume()
+
+  while (consume.stream.read() != null) {
+    // Loop
+  }
+}
+
+function consumeEnd (consume) {
+  const { type, body, resolve, stream, length } = consume
+
+  try {
+    if (type === 'text') {
+      resolve(toUSVString(Buffer.concat(body)))
+    } else if (type === 'json') {
+      resolve(JSON.parse(Buffer.concat(body)))
+    } else if (type === 'arrayBuffer') {
+      const dst = new Uint8Array(length)
+
+      let pos = 0
+      for (const buf of body) {
+        dst.set(buf, pos)
+        pos += buf.byteLength
+      }
+
+      resolve(dst.buffer)
+    } else if (type === 'blob') {
+      if (!Blob) {
+        Blob = require('buffer').Blob
+      }
+      resolve(new Blob(body, { type: stream[kContentType] }))
+    }
+
+    consumeFinish(consume)
+  } catch (err) {
+    stream.destroy(err)
+  }
+}
+
+function consumePush (consume, chunk) {
+  consume.length += chunk.length
+  consume.body.push(chunk)
+}
+
+function consumeFinish (consume, err) {
+  if (consume.body === null) {
+    return
+  }
+
+  if (err) {
+    consume.reject(err)
+  } else {
+    consume.resolve()
+  }
+
+  consume.type = null
+  consume.stream = null
+  consume.resolve = null
+  consume.reject = null
+  consume.length = 0
+  consume.body = null
+}
diff --git a/lib/api/util.js b/lib/api/util.js
new file mode 100644
index 0000000..bffd702
--- /dev/null
+++ b/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 }