basant307/AI_Governance_Project

1'use strict'2 3const {4  Readable,5  Duplex,6  PassThrough7} = require('node:stream')8const {9  InvalidArgumentError,10  InvalidReturnValueError,11  RequestAbortedError12} = require('../core/errors')13const util = require('../core/util')14const { AsyncResource } = require('node:async_hooks')15const { addSignal, removeSignal } = require('./abort-signal')16const assert = require('node:assert')17 18const kResume = Symbol('resume')19 20class PipelineRequest extends Readable {21  constructor () {22    super({ autoDestroy: true })23 24    this[kResume] = null25  }26 27  _read () {28    const { [kResume]: resume } = this29 30    if (resume) {31      this[kResume] = null32      resume()33    }34  }35 36  _destroy (err, callback) {37    this._read()38 39    callback(err)40  }41}42 43class PipelineResponse extends Readable {44  constructor (resume) {45    super({ autoDestroy: true })46    this[kResume] = resume47  }48 49  _read () {50    this[kResume]()51  }52 53  _destroy (err, callback) {54    if (!err && !this._readableState.endEmitted) {55      err = new RequestAbortedError()56    }57 58    callback(err)59  }60}61 62class PipelineHandler extends AsyncResource {63  constructor (opts, handler) {64    if (!opts || typeof opts !== 'object') {65      throw new InvalidArgumentError('invalid opts')66    }67 68    if (typeof handler !== 'function') {69      throw new InvalidArgumentError('invalid handler')70    }71 72    const { signal, method, opaque, onInfo, responseHeaders } = opts73 74    if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {75      throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget')76    }77 78    if (method === 'CONNECT') {79      throw new InvalidArgumentError('invalid method')80    }81 82    if (onInfo && typeof onInfo !== 'function') {83      throw new InvalidArgumentError('invalid onInfo callback')84    }85 86    super('UNDICI_PIPELINE')87 88    this.opaque = opaque || null89    this.responseHeaders = responseHeaders || null90    this.handler = handler91    this.abort = null92    this.context = null93    this.onInfo = onInfo || null94 95    this.req = new PipelineRequest().on('error', util.nop)96 97    this.ret = new Duplex({98      readableObjectMode: opts.objectMode,99      autoDestroy: true,100      read: () => {101        const { body } = this102 103        if (body?.resume) {104          body.resume()105        }106      },107      write: (chunk, encoding, callback) => {108        const { req } = this109 110        if (req.push(chunk, encoding) || req._readableState.destroyed) {111          callback()112        } else {113          req[kResume] = callback114        }115      },116      destroy: (err, callback) => {117        const { body, req, res, ret, abort } = this118 119        if (!err && !ret._readableState.endEmitted) {120          err = new RequestAbortedError()121        }122 123        if (abort && err) {124          abort()125        }126 127        util.destroy(body, err)128        util.destroy(req, err)129        util.destroy(res, err)130 131        removeSignal(this)132 133        callback(err)134      }135    }).on('prefinish', () => {136      const { req } = this137 138      // Node < 15 does not call _final in same tick.139      req.push(null)140    })141 142    this.res = null143 144    addSignal(this, signal)145  }146 147  onConnect (abort, context) {148    const { ret, res } = this149 150    if (this.reason) {151      abort(this.reason)152      return153    }154 155    assert(!res, 'pipeline cannot be retried')156    assert(!ret.destroyed)157 158    this.abort = abort159    this.context = context160  }161 162  onHeaders (statusCode, rawHeaders, resume) {163    const { opaque, handler, context } = this164 165    if (statusCode < 200) {166      if (this.onInfo) {167        const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)168        this.onInfo({ statusCode, headers })169      }170      return171    }172 173    this.res = new PipelineResponse(resume)174 175    let body176    try {177      this.handler = null178      const headers = this.responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)179      body = this.runInAsyncScope(handler, null, {180        statusCode,181        headers,182        opaque,183        body: this.res,184        context185      })186    } catch (err) {187      this.res.on('error', util.nop)188      throw err189    }190 191    if (!body || typeof body.on !== 'function') {192      throw new InvalidReturnValueError('expected Readable')193    }194 195    body196      .on('data', (chunk) => {197        const { ret, body } = this198 199        if (!ret.push(chunk) && body.pause) {200          body.pause()201        }202      })203      .on('error', (err) => {204        const { ret } = this205 206        util.destroy(ret, err)207      })208      .on('end', () => {209        const { ret } = this210 211        ret.push(null)212      })213      .on('close', () => {214        const { ret } = this215 216        if (!ret._readableState.ended) {217          util.destroy(ret, new RequestAbortedError())218        }219      })220 221    this.body = body222  }223 224  onData (chunk) {225    const { res } = this226    return res.push(chunk)227  }228 229  onComplete (trailers) {230    const { res } = this231    res.push(null)232  }233 234  onError (err) {235    const { ret } = this236    this.handler = null237    util.destroy(ret, err)238  }239}240 241function pipeline (opts, handler) {242  try {243    const pipelineHandler = new PipelineHandler(opts, handler)244    this.dispatch({ ...opts, body: pipelineHandler.req }, pipelineHandler)245    return pipelineHandler.ret246  } catch (err) {247    return new PassThrough().destroy(err)248  }249}250 251module.exports = pipeline252