{{ message }}
Repository navigation
Expand file tree
/
Copy pathlistener.ts
More file actions
316 lines (286 loc) · 10.2 KB
/
Copy pathlistener.ts
File metadata and controls
316 lines (286 loc) · 10.2 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
import type { IncomingMessage, ServerResponse, OutgoingHttpHeaders } from 'node:http'
import { Http2ServerRequest } from 'node:http2'
import type { Http2ServerResponse } from 'node:http2'
import type { Writable } from 'node:stream'
import type { IncomingMessageWithWrapBodyStream } from './request'
import {
abortControllerKey,
newRequest,
Request as LightweightRequest,
wrapBodyStream,
toRequestError,
} from './request'
import { cacheKey, Response as LightweightResponse } from './response'
import type { InternalCache } from './response'
import type { CustomErrorHandler, FetchCallback, HttpBindings } from './types'
import {
readWithoutBlocking,
writeFromReadableStream,
writeFromReadableStreamDefaultReader,
buildOutgoingHttpHeaders,
} from './utils'
import { X_ALREADY_SENT } from './utils/response/constants'
import './globals'
const outgoingEnded = Symbol('outgoingEnded')
type OutgoingHasOutgoingEnded = Http2ServerResponse & {
[outgoingEnded]?: () => void
}
const handleRequestError = (): Response =>
new Response(null, {
status: 400,
})
const handleFetchError = (e: unknown): Response =>
new Response(null, {
status:
e instanceof Error && (e.name === 'TimeoutError' || e.constructor.name === 'TimeoutError')
? 504 // timeout error emits 504 timeout
: 500,
})
const handleResponseError = (e: unknown, outgoing: ServerResponse | Http2ServerResponse) => {
const err = (e instanceof Error ? e : new Error('unknown error', { cause: e })) as Error & {
code: string
}
if (err.code === 'ERR_STREAM_PREMATURE_CLOSE') {
console.info('The user aborted a request.')
} else {
console.error(e)
if (!outgoing.headersSent) {
outgoing.writeHead(500, { 'Content-Type': 'text/plain' })
}
outgoing.end(`Error: ${err.message}`)
outgoing.destroy(err)
}
}
const flushHeaders = (outgoing: ServerResponse | Http2ServerResponse) => {
// If outgoing is ServerResponse (HTTP/1.1), it requires this to flush headers.
// However, Http2ServerResponse is sent without this.
if ('flushHeaders' in outgoing && outgoing.writable) {
outgoing.flushHeaders()
}
}
const responseViaCache = async (
res: Response,
outgoing: ServerResponse | Http2ServerResponse
): Promise<undefined | void> => {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
let [status, body, header] = (res as any)[cacheKey] as InternalCache
if (header instanceof Headers) {
header = buildOutgoingHttpHeaders(header)
}
// in `responseViaCache`, if body is not stream, Transfer-Encoding is considered not chunked
if (typeof body === 'string') {
header['Content-Length'] = Buffer.byteLength(body)
} else if (body instanceof Uint8Array) {
header['Content-Length'] = body.byteLength
} else if (body instanceof Blob) {
header['Content-Length'] = body.size
}
outgoing.writeHead(status, header)
if (typeof body === 'string' || body instanceof Uint8Array) {
outgoing.end(body)
} else if (body instanceof Blob) {
outgoing.end(new Uint8Array(await body.arrayBuffer()))
} else {
flushHeaders(outgoing)
await writeFromReadableStream(body, outgoing)?.catch(
(e) => handleResponseError(e, outgoing) as undefined
)
}
;(outgoing as OutgoingHasOutgoingEnded)[outgoingEnded]?.()
}
const isPromise = (res: Response | Promise<Response>): res is Promise<Response> =>
typeof (res as Promise<Response>).then === 'function'
const responseViaResponseObject = async (
res: Response | Promise<Response>,
outgoing: ServerResponse | Http2ServerResponse,
options: { errorHandler?: CustomErrorHandler } = {}
) => {
if (isPromise(res)) {
if (options.errorHandler) {
try {
res = await res
} catch (err) {
const errRes = await options.errorHandler(err)
if (!errRes) {
return
}
res = errRes
}
} else {
res = await res.catch(handleFetchError)
}
}
if (cacheKey in res) {
return responseViaCache(res as Response, outgoing)
}
const resHeaderRecord: OutgoingHttpHeaders = buildOutgoingHttpHeaders(res.headers)
if (res.body) {
const reader = res.body.getReader()
const values: Uint8Array[] = []
let done = false
let currentReadPromise: Promise<ReadableStreamReadResult<Uint8Array>> | undefined = undefined
if (resHeaderRecord['transfer-encoding'] !== 'chunked') {
// In the case of synchronous responses, usually a maximum of two (or three in special cases) readings is done
let maxReadCount = 2
for (let i = 0; i < maxReadCount; i++) {
currentReadPromise ||= reader.read()
const chunk = await readWithoutBlocking(currentReadPromise).catch((e) => {
console.error(e)
done = true
})
if (!chunk) {
if (i === 1) {
// XXX: In Node.js v24, some response bodies are not read all the way through until the next task queue,
// so wait a moment and retry. (e.g. new Blob([new Uint8Array(contents)]) )
await new Promise((resolve) => setTimeout(resolve))
maxReadCount = 3
continue
}
// Error occurred or currentReadPromise is not yet resolved.
// If an error occurs, immediately break the loop.
// If currentReadPromise is not yet resolved, pass it to writeFromReadableStreamDefaultReader.
break
}
currentReadPromise = undefined
if (chunk.value) {
values.push(chunk.value)
}
if (chunk.done) {
done = true
break
}
}
if (done && !('content-length' in resHeaderRecord)) {
resHeaderRecord['content-length'] = values.reduce((acc, value) => acc + value.length, 0)
}
}
outgoing.writeHead(res.status, resHeaderRecord)
values.forEach((value) => {
;(outgoing as Writable).write(value)
})
if (done) {
outgoing.end()
} else {
if (values.length === 0) {
flushHeaders(outgoing)
}
await writeFromReadableStreamDefaultReader(reader, outgoing, currentReadPromise)
}
} else if (resHeaderRecord[X_ALREADY_SENT]) {
// do nothing, the response has already been sent
} else {
outgoing.writeHead(res.status, resHeaderRecord)
outgoing.end()
}
;(outgoing as OutgoingHasOutgoingEnded)[outgoingEnded]?.()
}
export const getRequestListener = (
fetchCallback: FetchCallback,
options: {
hostname?: string
errorHandler?: CustomErrorHandler
overrideGlobalObjects?: boolean
autoCleanupIncoming?: boolean
} = {}
) => {
const autoCleanupIncoming = options.autoCleanupIncoming ?? true
if (options.overrideGlobalObjects !== false && global.Request !== LightweightRequest) {
Object.defineProperty(global, 'Request', {
value: LightweightRequest,
})
Object.defineProperty(global, 'Response', {
value: LightweightResponse,
})
}
return async (
incoming: IncomingMessage | Http2ServerRequest,
outgoing: ServerResponse | Http2ServerResponse
) => {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
let res, req: any
try {
// `fetchCallback()` requests a Request object, but global.Request is expensive to generate,
// so generate a pseudo Request object with only the minimum required information.
req = newRequest(incoming, options.hostname)
let incomingEnded =
!autoCleanupIncoming || incoming.method === 'GET' || incoming.method === 'HEAD'
if (!incomingEnded) {
;(incoming as IncomingMessageWithWrapBodyStream)[wrapBodyStream] = true
incoming.on('end', () => {
incomingEnded = true
})
if (incoming instanceof Http2ServerRequest) {
// a Http2ServerResponse instance requires additional processing on exit
// since outgoing.on('close') is not called even after outgoing.end() is called
// when the state is incomplete
;(outgoing as OutgoingHasOutgoingEnded)[outgoingEnded] = () => {
// incoming is not consumed to the end
if (!incomingEnded) {
setTimeout(() => {
// in the case of a simple POST request, the cleanup process may be done automatically
// and end is called at this point. At that point, nothing is done.
if (!incomingEnded) {
setTimeout(() => {
incoming.destroy()
// a Http2ServerResponse instance will not terminate without also calling outgoing.destroy()
outgoing.destroy()
})
}
})
}
}
}
}
// Detect if request was aborted.
outgoing.on('close', () => {
const abortController = req[abortControllerKey] as AbortController | undefined
if (abortController) {
if (incoming.errored) {
req[abortControllerKey].abort(incoming.errored.toString())
} else if (!outgoing.writableFinished) {
req[abortControllerKey].abort('Client connection prematurely closed.')
}
}
// incoming is not consumed to the end
if (!incomingEnded) {
setTimeout(() => {
// in the case of a simple POST request, the cleanup process may be done automatically
// and end is called at this point. At that point, nothing is done.
if (!incomingEnded) {
setTimeout(() => {
incoming.destroy()
})
}
})
}
})
res = fetchCallback(req, { incoming, outgoing } as HttpBindings) as
| Response
| Promise<Response>
if (cacheKey in res) {
// synchronous, cacheable response
return responseViaCache(res as Response, outgoing)
}
} catch (e: unknown) {
if (!res) {
if (options.errorHandler) {
res = await options.errorHandler(req ? e : toRequestError(e))
if (!res) {
return
}
} else if (!req) {
res = handleRequestError()
} else {
res = handleFetchError(e)
}
} else {
return handleResponseError(e, outgoing)
}
}
try {
return await responseViaResponseObject(res, outgoing, options)
} catch (e) {
return handleResponseError(e, outgoing)
}
}
}
You can’t perform that action at this time.