2022-02-18 20:43:43 +01:00
|
|
|
import type { IncomingMessage } from 'http'
|
2022-07-21 20:29:19 +02:00
|
|
|
import { PassThrough, Readable } from 'stream'
|
2022-02-18 20:43:43 +01:00
|
|
|
|
2022-07-21 20:29:19 +02:00
|
|
|
export function requestToBodyStream(
|
|
|
|
context: { ReadableStream: typeof ReadableStream },
|
2022-09-23 14:01:36 +02:00
|
|
|
KUint8Array: typeof Uint8Array,
|
2022-07-21 20:29:19 +02:00
|
|
|
stream: Readable
|
|
|
|
) {
|
|
|
|
return new context.ReadableStream({
|
2022-02-18 20:43:43 +01:00
|
|
|
start(controller) {
|
2022-09-23 14:01:36 +02:00
|
|
|
stream.on('data', (chunk) =>
|
|
|
|
controller.enqueue(new KUint8Array([...new Uint8Array(chunk)]))
|
|
|
|
)
|
2022-07-21 20:29:19 +02:00
|
|
|
stream.on('end', () => controller.close())
|
|
|
|
stream.on('error', (err) => controller.error(err))
|
2022-02-18 20:43:43 +01:00
|
|
|
},
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
|
|
|
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') {
|
2022-06-27 22:56:14 +02:00
|
|
|
v = v.bind(base)
|
2022-02-18 20:43:43 +01:00
|
|
|
}
|
|
|
|
base[key as keyof T] = v
|
|
|
|
}
|
|
|
|
|
|
|
|
return base
|
|
|
|
}
|
|
|
|
|
2022-07-21 20:29:19 +02:00
|
|
|
export interface ClonableBody {
|
|
|
|
finalize(): Promise<void>
|
|
|
|
cloneBodyStream(): Readable
|
|
|
|
}
|
|
|
|
|
|
|
|
export function getClonableBody<T extends IncomingMessage>(
|
|
|
|
readable: T
|
|
|
|
): ClonableBody {
|
|
|
|
let buffered: Readable | null = null
|
2022-02-18 20:43:43 +01:00
|
|
|
|
2022-07-28 09:50:51 +02:00
|
|
|
const endPromise = new Promise<void | { error?: unknown }>(
|
|
|
|
(resolve, reject) => {
|
|
|
|
readable.on('end', resolve)
|
|
|
|
readable.on('error', reject)
|
|
|
|
}
|
|
|
|
).catch((error) => {
|
|
|
|
return { error }
|
2022-03-08 20:24:58 +01:00
|
|
|
})
|
|
|
|
|
2022-02-18 20:43:43 +01:00
|
|
|
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.
|
|
|
|
*/
|
2022-03-08 20:24:58 +01:00
|
|
|
async finalize(): Promise<void> {
|
2022-07-21 20:29:19 +02:00
|
|
|
if (buffered) {
|
2022-07-28 09:50:51 +02:00
|
|
|
const res = await endPromise
|
|
|
|
|
|
|
|
if (res && typeof res === 'object' && res.error) {
|
|
|
|
throw res.error
|
|
|
|
}
|
2022-07-21 20:29:19 +02:00
|
|
|
replaceRequestBody(readable, buffered)
|
|
|
|
buffered = readable
|
2022-02-18 20:43:43 +01:00
|
|
|
}
|
|
|
|
},
|
|
|
|
/**
|
|
|
|
* Clones the body stream
|
|
|
|
* to pass into a middleware
|
|
|
|
*/
|
2022-07-21 20:29:19 +02:00
|
|
|
cloneBodyStream() {
|
|
|
|
const input = buffered ?? readable
|
|
|
|
const p1 = new PassThrough()
|
|
|
|
const p2 = new PassThrough()
|
2022-10-18 00:26:16 +02:00
|
|
|
input.on('data', (chunk) => {
|
|
|
|
p1.push(chunk)
|
|
|
|
p2.push(chunk)
|
|
|
|
})
|
|
|
|
input.on('end', () => {
|
|
|
|
p1.push(null)
|
|
|
|
p2.push(null)
|
|
|
|
})
|
2022-07-21 20:29:19 +02:00
|
|
|
buffered = p2
|
|
|
|
return p1
|
2022-02-18 20:43:43 +01:00
|
|
|
},
|
|
|
|
}
|
|
|
|
}
|