File Explorer

1'use strict'2 3const assert = require('node:assert')4const { finished, PassThrough } = require('node:stream')5const { InvalidArgumentError, InvalidReturnValueError } = require('../core/errors')6const util = require('../core/util')7const { getResolveErrorBodyCallback } = require('./util')8const { AsyncResource } = require('node:async_hooks')9const { addSignal, removeSignal } = require('./abort-signal')10 11class StreamHandler extends AsyncResource {12  constructor (opts, factory, callback) {13    if (!opts || typeof opts !== 'object') {14      throw new InvalidArgumentError('invalid opts')15    }16 17    const { signal, method, opaque, body, onInfo, responseHeaders, throwOnError } = opts18 19    try {20      if (typeof callback !== 'function') {21        throw new InvalidArgumentError('invalid callback')22      }23 24      if (typeof factory !== 'function') {25        throw new InvalidArgumentError('invalid factory')26      }27 28      if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {29        throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget')30      }31 32      if (method === 'CONNECT') {33        throw new InvalidArgumentError('invalid method')34      }35 36      if (onInfo && typeof onInfo !== 'function') {37        throw new InvalidArgumentError('invalid onInfo callback')38      }39 40      super('UNDICI_STREAM')41    } catch (err) {42      if (util.isStream(body)) {43        util.destroy(body.on('error', util.nop), err)44      }45      throw err46    }47 48    this.responseHeaders = responseHeaders || null49    this.opaque = opaque || null50    this.factory = factory51    this.callback = callback52    this.res = null53    this.abort = null54    this.context = null55    this.trailers = null56    this.body = body57    this.onInfo = onInfo || null58    this.throwOnError = throwOnError || false59 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, callback, 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    let res96 97    if (this.throwOnError && statusCode >= 400) {98      const parsedHeaders = responseHeaders === 'raw' ? util.parseHeaders(rawHeaders) : headers99      const contentType = parsedHeaders['content-type']100      res = new PassThrough()101 102      this.callback = null103      this.runInAsyncScope(getResolveErrorBodyCallback, null,104        { callback, body: res, contentType, statusCode, statusMessage, headers }105      )106    } else {107      if (factory === null) {108        return109      }110 111      res = this.runInAsyncScope(factory, null, {112        statusCode,113        headers,114        opaque,115        context116      })117 118      if (119        !res ||120        typeof res.write !== 'function' ||121        typeof res.end !== 'function' ||122        typeof res.on !== 'function'123      ) {124        throw new InvalidReturnValueError('expected Writable')125      }126 127      // TODO: Avoid finished. It registers an unnecessary amount of listeners.128      finished(res, { readable: false }, (err) => {129        const { callback, res, opaque, trailers, abort } = this130 131        this.res = null132        if (err || !res.readable) {133          util.destroy(res, err)134        }135 136        this.callback = null137        this.runInAsyncScope(callback, null, err || null, { opaque, trailers })138 139        if (err) {140          abort()141        }142      })143    }144 145    res.on('drain', resume)146 147    this.res = res148 149    const needDrain = res.writableNeedDrain !== undefined150      ? res.writableNeedDrain151      : res._writableState?.needDrain152 153    return needDrain !== true154  }155 156  onData (chunk) {157    const { res } = this158 159    return res ? res.write(chunk) : true160  }161 162  onComplete (trailers) {163    const { res } = this164 165    removeSignal(this)166 167    if (!res) {168      return169    }170 171    this.trailers = util.parseHeaders(trailers)172 173    res.end()174  }175 176  onError (err) {177    const { res, callback, opaque, body } = this178 179    removeSignal(this)180 181    this.factory = null182 183    if (res) {184      this.res = null185      util.destroy(res, err)186    } else if (callback) {187      this.callback = null188      queueMicrotask(() => {189        this.runInAsyncScope(callback, null, err, { opaque })190      })191    }192 193    if (body) {194      this.body = null195      util.destroy(body, err)196    }197  }198}199 200function stream (opts, factory, callback) {201  if (callback === undefined) {202    return new Promise((resolve, reject) => {203      stream.call(this, opts, factory, (err, data) => {204        return err ? reject(err) : resolve(data)205      })206    })207  }208 209  try {210    this.dispatch(opts, new StreamHandler(opts, factory, callback))211  } catch (err) {212    if (typeof callback !== 'function') {213      throw err214    }215    const opaque = opts?.opaque216    queueMicrotask(() => callback(err, { opaque }))217  }218}219 220module.exports = stream221