File Explorer

1'use strict'2 3const {4  Readable,5  Duplex,6  PassThrough7} = require('node:stream')8const assert = require('node:assert')9const { AsyncResource } = require('node:async_hooks')10const {11  InvalidArgumentError,12  InvalidReturnValueError,13  RequestAbortedError14} = require('../core/errors')15const util = require('../core/util')16const { addSignal, removeSignal } = require('./abort-signal')17 18function noop () {}19 20const kResume = Symbol('resume')21 22class PipelineRequest extends Readable {23  constructor () {24    super({ autoDestroy: true })25 26    this[kResume] = null27  }28 29  _read () {30    const { [kResume]: resume } = this31 32    if (resume) {33      this[kResume] = null34      resume()35    }36  }37 38  _destroy (err, callback) {39    this._read()40 41    callback(err)42  }43}44 45class PipelineResponse extends Readable {46  constructor (resume) {47    super({ autoDestroy: true })48    this[kResume] = resume49  }50 51  _read () {52    this[kResume]()53  }54 55  _destroy (err, callback) {56    if (!err && !this._readableState.endEmitted) {57      err = new RequestAbortedError()58    }59 60    callback(err)61  }62}63 64class PipelineHandler extends AsyncResource {65  constructor (opts, handler) {66    if (!opts || typeof opts !== 'object') {67      throw new InvalidArgumentError('invalid opts')68    }69 70    if (typeof handler !== 'function') {71      throw new InvalidArgumentError('invalid handler')72    }73 74    const { signal, method, opaque, onInfo, responseHeaders } = opts75 76    if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {77      throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget')78    }79 80    if (method === 'CONNECT') {81      throw new InvalidArgumentError('invalid method')82    }83 84    if (onInfo && typeof onInfo !== 'function') {85      throw new InvalidArgumentError('invalid onInfo callback')86    }87 88    super('UNDICI_PIPELINE')89 90    this.opaque = opaque || null91    this.responseHeaders = responseHeaders || null92    this.handler = handler93    this.abort = null94    this.context = null95    this.onInfo = onInfo || null96 97    this.req = new PipelineRequest().on('error', noop)98 99    this.ret = new Duplex({100      readableObjectMode: opts.objectMode,101      autoDestroy: true,102      read: () => {103        const { body } = this104 105        if (body?.resume) {106          body.resume()107        }108      },109      write: (chunk, encoding, callback) => {110        const { req } = this111 112        if (req.push(chunk, encoding) || req._readableState.destroyed) {113          callback()114        } else {115          req[kResume] = callback116        }117      },118      destroy: (err, callback) => {119        const { body, req, res, ret, abort } = this120 121        if (!err && !ret._readableState.endEmitted) {122          err = new RequestAbortedError()123        }124 125        if (abort && err) {126          abort()127        }128 129        util.destroy(body, err)130        util.destroy(req, err)131        util.destroy(res, err)132 133        removeSignal(this)134 135        callback(err)136      }137    }).on('prefinish', () => {138      const { req } = this139 140      // Node < 15 does not call _final in same tick.141      req.push(null)142    })143 144    this.res = null145 146    addSignal(this, signal)147  }148 149  onConnect (abort, context) {150    const { res } = this151 152    if (this.reason) {153      abort(this.reason)154      return155    }156 157    assert(!res, 'pipeline cannot be retried')158 159    this.abort = abort160    this.context = context161  }162 163  onHeaders (statusCode, rawHeaders, resume) {164    const { opaque, handler, context } = this165 166    if (statusCode < 200) {167      if (this.onInfo) {168        const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)169        this.onInfo({ statusCode, headers })170      }171      return172    }173 174    this.res = new PipelineResponse(resume)175 176    let body177    try {178      this.handler = null179      const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)180      body = this.runInAsyncScope(handler, null, {181        statusCode,182        headers,183        opaque,184        body: this.res,185        context186      })187    } catch (err) {188      this.res.on('error', noop)189      throw err190    }191 192    if (!body || typeof body.on !== 'function') {193      throw new InvalidReturnValueError('expected Readable')194    }195 196    body197      .on('data', (chunk) => {198        const { ret, body } = this199 200        if (!ret.push(chunk) && body.pause) {201          body.pause()202        }203      })204      .on('error', (err) => {205        const { ret } = this206 207        util.destroy(ret, err)208      })209      .on('end', () => {210        const { ret } = this211 212        ret.push(null)213      })214      .on('close', () => {215        const { ret } = this216 217        if (!ret._readableState.ended) {218          util.destroy(ret, new RequestAbortedError())219        }220      })221 222    this.body = body223  }224 225  onData (chunk) {226    const { res } = this227    return res.push(chunk)228  }229 230  onComplete (trailers) {231    const { res } = this232    res.push(null)233  }234 235  onError (err) {236    const { ret } = this237    this.handler = null238    util.destroy(ret, err)239  }240}241 242function pipeline (opts, handler) {243  try {244    const pipelineHandler = new PipelineHandler(opts, handler)245    this.dispatch({ ...opts, body: pipelineHandler.req }, pipelineHandler)246    return pipelineHandler.ret247  } catch (err) {248    return new PassThrough().destroy(err)249  }250}251 252module.exports = pipeline253