File Explorer

1'use strict'2 3const assert = require('node:assert')4const { Readable } = require('node:stream')5const { RequestAbortedError, NotSupportedError, InvalidArgumentError, AbortError } = require('../core/errors')6const util = require('../core/util')7const { ReadableStreamFrom } = require('../core/util')8 9const kConsume = Symbol('kConsume')10const kReading = Symbol('kReading')11const kBody = Symbol('kBody')12const kAbort = Symbol('kAbort')13const kContentType = Symbol('kContentType')14const kContentLength = Symbol('kContentLength')15const kUsed = Symbol('kUsed')16const kBytesRead = Symbol('kBytesRead')17 18const noop = () => {}19 20/**21 * @class22 * @extends {Readable}23 * @see https://fetch.spec.whatwg.org/#body24 */25class BodyReadable extends Readable {26  /**27   * @param {object} opts28   * @param {(this: Readable, size: number) => void} opts.resume29   * @param {() => (void | null)} opts.abort30   * @param {string} [opts.contentType = '']31   * @param {number} [opts.contentLength]32   * @param {number} [opts.highWaterMark = 64 * 1024]33   */34  constructor ({35    resume,36    abort,37    contentType = '',38    contentLength,39    highWaterMark = 64 * 1024 // Same as nodejs fs streams.40  }) {41    super({42      autoDestroy: true,43      read: resume,44      highWaterMark45    })46 47    this._readableState.dataEmitted = false48 49    this[kAbort] = abort50 51    /** @type {Consume | null} */52    this[kConsume] = null53 54    /** @type {number} */55    this[kBytesRead] = 056 57    /** @type {ReadableStream|null} */58    this[kBody] = null59 60    /** @type {boolean} */61    this[kUsed] = false62 63    /** @type {string} */64    this[kContentType] = contentType65 66    /** @type {number|null} */67    this[kContentLength] = Number.isFinite(contentLength) ? contentLength : null68 69    /**70     * Is stream being consumed through Readable API?71     * This is an optimization so that we avoid checking72     * for 'data' and 'readable' listeners in the hot path73     * inside push().74     *75     * @type {boolean}76     */77    this[kReading] = false78  }79 80  /**81   * @param {Error|null} err82   * @param {(error:(Error|null)) => void} callback83   * @returns {void}84   */85  _destroy (err, callback) {86    if (!err && !this._readableState.endEmitted) {87      err = new RequestAbortedError()88    }89 90    if (err) {91      this[kAbort]()92    }93 94    // Workaround for Node "bug". If the stream is destroyed in same95    // tick as it is created, then a user who is waiting for a96    // promise (i.e micro tick) for installing an 'error' listener will97    // never get a chance and will always encounter an unhandled exception.98    if (!this[kUsed]) {99      setImmediate(callback, err)100    } else {101      callback(err)102    }103  }104 105  /**106   * @param {string|symbol} event107   * @param {(...args: any[]) => void} listener108   * @returns {this}109   */110  on (event, listener) {111    if (event === 'data' || event === 'readable') {112      this[kReading] = true113      this[kUsed] = true114    }115    return super.on(event, listener)116  }117 118  /**119   * @param {string|symbol} event120   * @param {(...args: any[]) => void} listener121   * @returns {this}122   */123  addListener (event, listener) {124    return this.on(event, listener)125  }126 127  /**128   * @param {string|symbol} event129   * @param {(...args: any[]) => void} listener130   * @returns {this}131   */132  off (event, listener) {133    const ret = super.off(event, listener)134    if (event === 'data' || event === 'readable') {135      this[kReading] = (136        this.listenerCount('data') > 0 ||137        this.listenerCount('readable') > 0138      )139    }140    return ret141  }142 143  /**144   * @param {string|symbol} event145   * @param {(...args: any[]) => void} listener146   * @returns {this}147   */148  removeListener (event, listener) {149    return this.off(event, listener)150  }151 152  /**153   * @param {Buffer|null} chunk154   * @returns {boolean}155   */156  push (chunk) {157    if (chunk) {158      this[kBytesRead] += chunk.length159      if (this[kConsume]) {160        consumePush(this[kConsume], chunk)161        return this[kReading] ? super.push(chunk) : true162      }163    }164 165    return super.push(chunk)166  }167 168  /**169   * Consumes and returns the body as a string.170   *171   * @see https://fetch.spec.whatwg.org/#dom-body-text172   * @returns {Promise<string>}173   */174  text () {175    return consume(this, 'text')176  }177 178  /**179   * Consumes and returns the body as a JavaScript Object.180   *181   * @see https://fetch.spec.whatwg.org/#dom-body-json182   * @returns {Promise<unknown>}183   */184  json () {185    return consume(this, 'json')186  }187 188  /**189   * Consumes and returns the body as a Blob190   *191   * @see https://fetch.spec.whatwg.org/#dom-body-blob192   * @returns {Promise<Blob>}193   */194  blob () {195    return consume(this, 'blob')196  }197 198  /**199   * Consumes and returns the body as an Uint8Array.200   *201   * @see https://fetch.spec.whatwg.org/#dom-body-bytes202   * @returns {Promise<Uint8Array>}203   */204  bytes () {205    return consume(this, 'bytes')206  }207 208  /**209   * Consumes and returns the body as an ArrayBuffer.210   *211   * @see https://fetch.spec.whatwg.org/#dom-body-arraybuffer212   * @returns {Promise<ArrayBuffer>}213   */214  arrayBuffer () {215    return consume(this, 'arrayBuffer')216  }217 218  /**219   * Not implemented220   *221   * @see https://fetch.spec.whatwg.org/#dom-body-formdata222   * @throws {NotSupportedError}223   */224  async formData () {225    // TODO: Implement.226    throw new NotSupportedError()227  }228 229  /**230   * Returns true if the body is not null and the body has been consumed.231   * Otherwise, returns false.232   *233   * @see https://fetch.spec.whatwg.org/#dom-body-bodyused234   * @readonly235   * @returns {boolean}236   */237  get bodyUsed () {238    return util.isDisturbed(this)239  }240 241  /**242   * @see https://fetch.spec.whatwg.org/#dom-body-body243   * @readonly244   * @returns {ReadableStream}245   */246  get body () {247    if (!this[kBody]) {248      this[kBody] = ReadableStreamFrom(this)249      if (this[kConsume]) {250        // TODO: Is this the best way to force a lock?251        this[kBody].getReader() // Ensure stream is locked.252        assert(this[kBody].locked)253      }254    }255    return this[kBody]256  }257 258  /**259   * Dumps the response body by reading `limit` number of bytes.260   * @param {object} opts261   * @param {number} [opts.limit = 131072] Number of bytes to read.262   * @param {AbortSignal} [opts.signal] An AbortSignal to cancel the dump.263   * @returns {Promise<null>}264   */265  dump (opts) {266    const signal = opts?.signal267 268    if (signal != null && (typeof signal !== 'object' || !('aborted' in signal))) {269      return Promise.reject(new InvalidArgumentError('signal must be an AbortSignal'))270    }271 272    const limit = opts?.limit && Number.isFinite(opts.limit)273      ? opts.limit274      : 128 * 1024275 276    if (signal?.aborted) {277      return Promise.reject(signal.reason ?? new AbortError())278    }279 280    if (this._readableState.closeEmitted) {281      return Promise.resolve(null)282    }283 284    return new Promise((resolve, reject) => {285      if (286        (this[kContentLength] && (this[kContentLength] > limit)) ||287        this[kBytesRead] > limit288      ) {289        this.destroy(new AbortError())290      }291 292      if (signal) {293        const onAbort = () => {294          this.destroy(signal.reason ?? new AbortError())295        }296        signal.addEventListener('abort', onAbort)297        this298          .on('close', function () {299            signal.removeEventListener('abort', onAbort)300            if (signal.aborted) {301              reject(signal.reason ?? new AbortError())302            } else {303              resolve(null)304            }305          })306      } else {307        this.on('close', resolve)308      }309 310      this311        .on('error', noop)312        .on('data', () => {313          if (this[kBytesRead] > limit) {314            this.destroy()315          }316        })317        .resume()318    })319  }320 321  /**322   * @param {BufferEncoding} encoding323   * @returns {this}324   */325  setEncoding (encoding) {326    if (Buffer.isEncoding(encoding)) {327      this._readableState.encoding = encoding328    }329    return this330  }331}332 333/**334 * @see https://streams.spec.whatwg.org/#readablestream-locked335 * @param {BodyReadable} bodyReadable336 * @returns {boolean}337 */338function isLocked (bodyReadable) {339  // Consume is an implicit lock.340  return bodyReadable[kBody]?.locked === true || bodyReadable[kConsume] !== null341}342 343/**344 * @see https://fetch.spec.whatwg.org/#body-unusable345 * @param {BodyReadable} bodyReadable346 * @returns {boolean}347 */348function isUnusable (bodyReadable) {349  return util.isDisturbed(bodyReadable) || isLocked(bodyReadable)350}351 352/**353 * @typedef {'text' | 'json' | 'blob' | 'bytes' | 'arrayBuffer'} ConsumeType354 */355 356/**357 * @template {ConsumeType} T358 * @typedef {T extends 'text' ? string :359 *           T extends 'json' ? unknown :360 *           T extends 'blob' ? Blob :361 *           T extends 'arrayBuffer' ? ArrayBuffer :362 *           T extends 'bytes' ? Uint8Array :363 *           never364 * } ConsumeReturnType365 */366/**367 * @typedef {object} Consume368 * @property {ConsumeType} type369 * @property {BodyReadable} stream370 * @property {((value?: any) => void)} resolve371 * @property {((err: Error) => void)} reject372 * @property {number} length373 * @property {Buffer[]} body374 */375 376/**377 * @template {ConsumeType} T378 * @param {BodyReadable} stream379 * @param {T} type380 * @returns {Promise<ConsumeReturnType<T>>}381 */382function consume (stream, type) {383  assert(!stream[kConsume])384 385  return new Promise((resolve, reject) => {386    if (isUnusable(stream)) {387      const rState = stream._readableState388      if (rState.destroyed && rState.closeEmitted === false) {389        stream390          .on('error', reject)391          .on('close', () => {392            reject(new TypeError('unusable'))393          })394      } else {395        reject(rState.errored ?? new TypeError('unusable'))396      }397    } else {398      queueMicrotask(() => {399        stream[kConsume] = {400          type,401          stream,402          resolve,403          reject,404          length: 0,405          body: []406        }407 408        stream409          .on('error', function (err) {410            consumeFinish(this[kConsume], err)411          })412          .on('close', function () {413            if (this[kConsume].body !== null) {414              consumeFinish(this[kConsume], new RequestAbortedError())415            }416          })417 418        consumeStart(stream[kConsume])419      })420    }421  })422}423 424/**425 * @param {Consume} consume426 * @returns {void}427 */428function consumeStart (consume) {429  if (consume.body === null) {430    return431  }432 433  const { _readableState: state } = consume.stream434 435  if (state.bufferIndex) {436    const start = state.bufferIndex437    const end = state.buffer.length438    for (let n = start; n < end; n++) {439      consumePush(consume, state.buffer[n])440    }441  } else {442    for (const chunk of state.buffer) {443      consumePush(consume, chunk)444    }445  }446 447  if (state.endEmitted) {448    consumeEnd(this[kConsume], this._readableState.encoding)449  } else {450    consume.stream.on('end', function () {451      consumeEnd(this[kConsume], this._readableState.encoding)452    })453  }454 455  consume.stream.resume()456 457  while (consume.stream.read() != null) {458    // Loop459  }460}461 462/**463 * @param {Buffer[]} chunks464 * @param {number} length465 * @param {BufferEncoding} [encoding='utf8']466 * @returns {string}467 */468function chunksDecode (chunks, length, encoding) {469  if (chunks.length === 0 || length === 0) {470    return ''471  }472  const buffer = chunks.length === 1 ? chunks[0] : Buffer.concat(chunks, length)473  const bufferLength = buffer.length474 475  // Skip BOM.476  const start =477    bufferLength > 2 &&478    buffer[0] === 0xef &&479    buffer[1] === 0xbb &&480    buffer[2] === 0xbf481      ? 3482      : 0483  if (!encoding || encoding === 'utf8' || encoding === 'utf-8') {484    return buffer.utf8Slice(start, bufferLength)485  } else {486    return buffer.subarray(start, bufferLength).toString(encoding)487  }488}489 490/**491 * @param {Buffer[]} chunks492 * @param {number} length493 * @returns {Uint8Array}494 */495function chunksConcat (chunks, length) {496  if (chunks.length === 0 || length === 0) {497    return new Uint8Array(0)498  }499  if (chunks.length === 1) {500    // fast-path501    return new Uint8Array(chunks[0])502  }503  const buffer = new Uint8Array(Buffer.allocUnsafeSlow(length).buffer)504 505  let offset = 0506  for (let i = 0; i < chunks.length; ++i) {507    const chunk = chunks[i]508    buffer.set(chunk, offset)509    offset += chunk.length510  }511 512  return buffer513}514 515/**516 * @param {Consume} consume517 * @param {BufferEncoding} encoding518 * @returns {void}519 */520function consumeEnd (consume, encoding) {521  const { type, body, resolve, stream, length } = consume522 523  try {524    if (type === 'text') {525      resolve(chunksDecode(body, length, encoding))526    } else if (type === 'json') {527      resolve(JSON.parse(chunksDecode(body, length, encoding)))528    } else if (type === 'arrayBuffer') {529      resolve(chunksConcat(body, length).buffer)530    } else if (type === 'blob') {531      resolve(new Blob(body, { type: stream[kContentType] }))532    } else if (type === 'bytes') {533      resolve(chunksConcat(body, length))534    }535 536    consumeFinish(consume)537  } catch (err) {538    stream.destroy(err)539  }540}541 542/**543 * @param {Consume} consume544 * @param {Buffer} chunk545 * @returns {void}546 */547function consumePush (consume, chunk) {548  consume.length += chunk.length549  consume.body.push(chunk)550}551 552/**553 * @param {Consume} consume554 * @param {Error} [err]555 * @returns {void}556 */557function consumeFinish (consume, err) {558  if (consume.body === null) {559    return560  }561 562  if (err) {563    consume.reject(err)564  } else {565    consume.resolve()566  }567 568  // Reset the consume object to allow for garbage collection.569  consume.type = null570  consume.stream = null571  consume.resolve = null572  consume.reject = null573  consume.length = 0574  consume.body = null575}576 577module.exports = {578  Readable: BodyReadable,579  chunksDecode580}581