codekingpro/portable-devtools

1'use strict'
2
3const {
4  Readable,
5  Duplex,
6  PassThrough
7} = require('node:stream')
8const {
9  InvalidArgumentError,
10  InvalidReturnValueError,
11  RequestAbortedError
12} = 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] = null
25  }
26
27  _read () {
28    const { [kResume]: resume } = this
29
30    if (resume) {
31      this[kResume] = null
32      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] = resume
47  }
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 } = opts
73
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 || null
89    this.responseHeaders = responseHeaders || null
90    this.handler = handler
91    this.abort = null
92    this.context = null
93    this.onInfo = onInfo || null
94
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 } = this
102
103        if (body?.resume) {
104          body.resume()
105        }
106      },
107      write: (chunk, encoding, callback) => {
108        const { req } = this
109
110        if (req.push(chunk, encoding) || req._readableState.destroyed) {
111          callback()
112        } else {
113          req[kResume] = callback
114        }
115      },
116      destroy: (err, callback) => {
117        const { body, req, res, ret, abort } = this
118
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 } = this
137
138      // Node < 15 does not call _final in same tick.
139      req.push(null)
140    })
141
142    this.res = null
143
144    addSignal(this, signal)
145  }
146
147  onConnect (abort, context) {
148    const { ret, res } = this
149
150    if (this.reason) {
151      abort(this.reason)
152      return
153    }
154
155    assert(!res, 'pipeline cannot be retried')
156    assert(!ret.destroyed)
157
158    this.abort = abort
159    this.context = context
160  }
161
162  onHeaders (statusCode, rawHeaders, resume) {
163    const { opaque, handler, context } = this
164
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      return
171    }
172
173    this.res = new PipelineResponse(resume)
174
175    let body
176    try {
177      this.handler = null
178      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        context
185      })
186    } catch (err) {
187      this.res.on('error', util.nop)
188      throw err
189    }
190
191    if (!body || typeof body.on !== 'function') {
192      throw new InvalidReturnValueError('expected Readable')
193    }
194
195    body
196      .on('data', (chunk) => {
197        const { ret, body } = this
198
199        if (!ret.push(chunk) && body.pause) {
200          body.pause()
201        }
202      })
203      .on('error', (err) => {
204        const { ret } = this
205
206        util.destroy(ret, err)
207      })
208      .on('end', () => {
209        const { ret } = this
210
211        ret.push(null)
212      })
213      .on('close', () => {
214        const { ret } = this
215
216        if (!ret._readableState.ended) {
217          util.destroy(ret, new RequestAbortedError())
218        }
219      })
220
221    this.body = body
222  }
223
224  onData (chunk) {
225    const { res } = this
226    return res.push(chunk)
227  }
228
229  onComplete (trailers) {
230    const { res } = this
231    res.push(null)
232  }
233
234  onError (err) {
235    const { ret } = this
236    this.handler = null
237    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.ret
246  } catch (err) {
247    return new PassThrough().destroy(err)
248  }
249}
250
251module.exports = pipeline
252