File Explorer

1'use strict'2 3const assert = require('node:assert')4const { finished } = require('node:stream')5const { AsyncResource } = require('node:async_hooks')6const { InvalidArgumentError, InvalidReturnValueError } = require('../core/errors')7const util = require('../core/util')8const { addSignal, removeSignal } = require('./abort-signal')9 10function noop () {}11 12class StreamHandler extends AsyncResource {13  constructor (opts, factory, callback) {14    if (!opts || typeof opts !== 'object') {15      throw new InvalidArgumentError('invalid opts')16    }17 18    const { signal, method, opaque, body, onInfo, responseHeaders } = opts19 20    try {21      if (typeof callback !== 'function') {22        throw new InvalidArgumentError('invalid callback')23      }24 25      if (typeof factory !== 'function') {26        throw new InvalidArgumentError('invalid factory')27      }28 29      if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {30        throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget')31      }32 33      if (method === 'CONNECT') {34        throw new InvalidArgumentError('invalid method')35      }36 37      if (onInfo && typeof onInfo !== 'function') {38        throw new InvalidArgumentError('invalid onInfo callback')39      }40 41      super('UNDICI_STREAM')42    } catch (err) {43      if (util.isStream(body)) {44        util.destroy(body.on('error', noop), err)45      }46      throw err47    }48 49    this.responseHeaders = responseHeaders || null50    this.opaque = opaque || null51    this.factory = factory52    this.callback = callback53    this.res = null54    this.abort = null55    this.context = null56    this.trailers = null57    this.body = body58    this.onInfo = onInfo || null59 60    if (util.isStream(body)) {61      body.on('error', (err) => {62        this.onError(err)63      })64    }65 66    addSignal(this, signal)67  }68 69  onConnect (abort, context) {70    if (this.reason) {71      abort(this.reason)72      return73    }74 75    assert(this.callback)76 77    this.abort = abort78    this.context = context79  }80 81  onHeaders (statusCode, rawHeaders, resume, statusMessage) {82    const { factory, opaque, context, responseHeaders } = this83 84    const headers = responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)85 86    if (statusCode < 200) {87      if (this.onInfo) {88        this.onInfo({ statusCode, headers })89      }90      return91    }92 93    this.factory = null94 95    if (factory === null) {96      return97    }98 99    const res = this.runInAsyncScope(factory, null, {100      statusCode,101      headers,102      opaque,103      context104    })105 106    if (107      !res ||108      typeof res.write !== 'function' ||109      typeof res.end !== 'function' ||110      typeof res.on !== 'function'111    ) {112      throw new InvalidReturnValueError('expected Writable')113    }114 115    // TODO: Avoid finished. It registers an unnecessary amount of listeners.116    finished(res, { readable: false }, (err) => {117      const { callback, res, opaque, trailers, abort } = this118 119      this.res = null120      if (err || !res?.readable) {121        util.destroy(res, err)122      }123 124      this.callback = null125      this.runInAsyncScope(callback, null, err || null, { opaque, trailers })126 127      if (err) {128        abort()129      }130    })131 132    res.on('drain', resume)133 134    this.res = res135 136    const needDrain = res.writableNeedDrain !== undefined137      ? res.writableNeedDrain138      : res._writableState?.needDrain139 140    return needDrain !== true141  }142 143  onData (chunk) {144    const { res } = this145 146    return res ? res.write(chunk) : true147  }148 149  onComplete (trailers) {150    const { res } = this151 152    removeSignal(this)153 154    if (!res) {155      return156    }157 158    this.trailers = util.parseHeaders(trailers)159 160    res.end()161  }162 163  onError (err) {164    const { res, callback, opaque, body } = this165 166    removeSignal(this)167 168    this.factory = null169 170    if (res) {171      this.res = null172      util.destroy(res, err)173    } else if (callback) {174      this.callback = null175      queueMicrotask(() => {176        this.runInAsyncScope(callback, null, err, { opaque })177      })178    }179 180    if (body) {181      this.body = null182      util.destroy(body, err)183    }184  }185}186 187function stream (opts, factory, callback) {188  if (callback === undefined) {189    return new Promise((resolve, reject) => {190      stream.call(this, opts, factory, (err, data) => {191        return err ? reject(err) : resolve(data)192      })193    })194  }195 196  try {197    const handler = new StreamHandler(opts, factory, callback)198 199    this.dispatch(opts, handler)200  } catch (err) {201    if (typeof callback !== 'function') {202      throw err203    }204    const opaque = opts?.opaque205    queueMicrotask(() => callback(err, { opaque }))206  }207}208 209module.exports = stream210