codekingpro/portable-devtools

1'use strict'
2
3const assert = require('node:assert')
4const { Readable } = require('./readable')
5const { InvalidArgumentError, RequestAbortedError } = require('../core/errors')
6const util = require('../core/util')
7const { getResolveErrorBodyCallback } = require('./util')
8const { AsyncResource } = require('node:async_hooks')
9
10class RequestHandler extends AsyncResource {
11  constructor (opts, callback) {
12    if (!opts || typeof opts !== 'object') {
13      throw new InvalidArgumentError('invalid opts')
14    }
15
16    const { signal, method, opaque, body, onInfo, responseHeaders, throwOnError, highWaterMark } = opts
17
18    try {
19      if (typeof callback !== 'function') {
20        throw new InvalidArgumentError('invalid callback')
21      }
22
23      if (highWaterMark && (typeof highWaterMark !== 'number' || highWaterMark < 0)) {
24        throw new InvalidArgumentError('invalid highWaterMark')
25      }
26
27      if (signal && typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') {
28        throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget')
29      }
30
31      if (method === 'CONNECT') {
32        throw new InvalidArgumentError('invalid method')
33      }
34
35      if (onInfo && typeof onInfo !== 'function') {
36        throw new InvalidArgumentError('invalid onInfo callback')
37      }
38
39      super('UNDICI_REQUEST')
40    } catch (err) {
41      if (util.isStream(body)) {
42        util.destroy(body.on('error', util.nop), err)
43      }
44      throw err
45    }
46
47    this.method = method
48    this.responseHeaders = responseHeaders || null
49    this.opaque = opaque || null
50    this.callback = callback
51    this.res = null
52    this.abort = null
53    this.body = body
54    this.trailers = {}
55    this.context = null
56    this.onInfo = onInfo || null
57    this.throwOnError = throwOnError
58    this.highWaterMark = highWaterMark
59    this.signal = signal
60    this.reason = null
61    this.removeAbortListener = null
62
63    if (util.isStream(body)) {
64      body.on('error', (err) => {
65        this.onError(err)
66      })
67    }
68
69    if (this.signal) {
70      if (this.signal.aborted) {
71        this.reason = this.signal.reason ?? new RequestAbortedError()
72      } else {
73        this.removeAbortListener = util.addAbortListener(this.signal, () => {
74          this.reason = this.signal.reason ?? new RequestAbortedError()
75          if (this.res) {
76            util.destroy(this.res.on('error', util.nop), this.reason)
77          } else if (this.abort) {
78            this.abort(this.reason)
79          }
80
81          if (this.removeAbortListener) {
82            this.res?.off('close', this.removeAbortListener)
83            this.removeAbortListener()
84            this.removeAbortListener = null
85          }
86        })
87      }
88    }
89  }
90
91  onConnect (abort, context) {
92    if (this.reason) {
93      abort(this.reason)
94      return
95    }
96
97    assert(this.callback)
98
99    this.abort = abort
100    this.context = context
101  }
102
103  onHeaders (statusCode, rawHeaders, resume, statusMessage) {
104    const { callback, opaque, abort, context, responseHeaders, highWaterMark } = this
105
106    const headers = responseHeaders === 'raw' ? util.parseRawHeaders(rawHeaders) : util.parseHeaders(rawHeaders)
107
108    if (statusCode < 200) {
109      if (this.onInfo) {
110        this.onInfo({ statusCode, headers })
111      }
112      return
113    }
114
115    const parsedHeaders = responseHeaders === 'raw' ? util.parseHeaders(rawHeaders) : headers
116    const contentType = parsedHeaders['content-type']
117    const contentLength = parsedHeaders['content-length']
118    const res = new Readable({
119      resume,
120      abort,
121      contentType,
122      contentLength: this.method !== 'HEAD' && contentLength
123        ? Number(contentLength)
124        : null,
125      highWaterMark
126    })
127
128    if (this.removeAbortListener) {
129      res.on('close', this.removeAbortListener)
130    }
131
132    this.callback = null
133    this.res = res
134    if (callback !== null) {
135      if (this.throwOnError && statusCode >= 400) {
136        this.runInAsyncScope(getResolveErrorBodyCallback, null,
137          { callback, body: res, contentType, statusCode, statusMessage, headers }
138        )
139      } else {
140        this.runInAsyncScope(callback, null, null, {
141          statusCode,
142          headers,
143          trailers: this.trailers,
144          opaque,
145          body: res,
146          context
147        })
148      }
149    }
150  }
151
152  onData (chunk) {
153    return this.res.push(chunk)
154  }
155
156  onComplete (trailers) {
157    util.parseHeaders(trailers, this.trailers)
158    this.res.push(null)
159  }
160
161  onError (err) {
162    const { res, callback, body, opaque } = this
163
164    if (callback) {
165      // TODO: Does this need queueMicrotask?
166      this.callback = null
167      queueMicrotask(() => {
168        this.runInAsyncScope(callback, null, err, { opaque })
169      })
170    }
171
172    if (res) {
173      this.res = null
174      // Ensure all queued handlers are invoked before destroying res.
175      queueMicrotask(() => {
176        util.destroy(res, err)
177      })
178    }
179
180    if (body) {
181      this.body = null
182      util.destroy(body, err)
183    }
184
185    if (this.removeAbortListener) {
186      res?.off('close', this.removeAbortListener)
187      this.removeAbortListener()
188      this.removeAbortListener = null
189    }
190  }
191}
192
193function request (opts, callback) {
194  if (callback === undefined) {
195    return new Promise((resolve, reject) => {
196      request.call(this, opts, (err, data) => {
197        return err ? reject(err) : resolve(data)
198      })
199    })
200  }
201
202  try {
203    this.dispatch(opts, new RequestHandler(opts, callback))
204  } catch (err) {
205    if (typeof callback !== 'function') {
206      throw err
207    }
208    const opaque = opts?.opaque
209    queueMicrotask(() => callback(err, { opaque }))
210  }
211}
212
213module.exports = request
214module.exports.RequestHandler = RequestHandler
215