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 },
|
|
|
|
stream: Readable
|
|
|
|
) {
|
|
|
|
return new context.ReadableStream({
|
2022-02-18 20:43:43 +01:00
|
|
|
start(controller) {
|
2022-07-21 20:29:19 +02:00
|
|
|
stream.on('data', (chunk) => controller.enqueue(chunk))
|
|
|
|
stream.on('end', () => controller.close())
|
|
|
|
stream.on('error', (err) => controller.error(err))
|
2022-02-18 20:43:43 +01:00
|
|
|
},
|
|
|
|
})
|
|
|
|
}
|
|
|
|
|
2022-07-21 20:29:19 +02:00
|
|
|
export function bodyStreamToNodeStream(
|
|
|
|
bodyStream: ReadableStream<Uint8Array>
|
|
|
|
): Readable {
|
2022-02-18 20:43:43 +01:00
|
|
|
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') {
|
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-03-08 20:24:58 +01:00
|
|
|
const endPromise = new Promise((resolve, reject) => {
|
2022-07-21 20:29:19 +02:00
|
|
|
readable.on('end', resolve)
|
|
|
|
readable.on('error', reject)
|
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-03-08 20:24:58 +01:00
|
|
|
await endPromise
|
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()
|
|
|
|
input.pipe(p1)
|
|
|
|
input.pipe(p2)
|
|
|
|
buffered = p2
|
|
|
|
return p1
|
2022-02-18 20:43:43 +01:00
|
|
|
},
|
|
|
|
}
|
|
|
|
}
|