-
Notifications
You must be signed in to change notification settings - Fork 26k
/
body-streams.ts
99 lines (90 loc) · 2.25 KB
/
body-streams.ts
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
import type { IncomingMessage } from 'http'
import { PassThrough, Readable } from 'stream'
export function requestToBodyStream(
context: { ReadableStream: typeof ReadableStream },
stream: Readable
) {
return new context.ReadableStream({
start(controller) {
stream.on('data', (chunk) => controller.enqueue(chunk))
stream.on('end', () => controller.close())
stream.on('error', (err) => controller.error(err))
},
})
}
export function bodyStreamToNodeStream(
bodyStream: ReadableStream<Uint8Array>
): Readable {
const reader = bodyStream.getReader()
return Readable.from(
(async function* () {
while (true) {
const { done, value } = await reader.read()
if (done) {
return
}
yield value
}
})()
)
}
function replaceRequestBody<T extends IncomingMessage>(
base: T,
stream: Readable
): T {
for (const key in stream) {
let v = stream[key as keyof Readable] as any
if (typeof v === 'function') {
v = v.bind(base)
}
base[key as keyof T] = v
}
return base
}
export interface ClonableBody {
finalize(): Promise<void>
cloneBodyStream(): Readable
}
export function getClonableBody<T extends IncomingMessage>(
readable: T
): ClonableBody {
let buffered: Readable | null = null
const endPromise = new Promise<void | { error?: unknown }>(
(resolve, reject) => {
readable.on('end', resolve)
readable.on('error', reject)
}
).catch((error) => {
return { error }
})
return {
/**
* Replaces the original request body if necessary.
* This is done because once we read the body from the original request,
* we can't read it again.
*/
async finalize(): Promise<void> {
if (buffered) {
const res = await endPromise
if (res && typeof res === 'object' && res.error) {
throw res.error
}
replaceRequestBody(readable, buffered)
buffered = readable
}
},
/**
* Clones the body stream
* to pass into a middleware
*/
cloneBodyStream() {
const input = buffered ?? readable
const p1 = new PassThrough()
const p2 = new PassThrough()
input.pipe(p1)
input.pipe(p2)
buffered = p2
return p1
},
}
}