133 lines
3.6 KiB
TypeScript
133 lines
3.6 KiB
TypeScript
/** @jest-environment node */
|
|
|
|
import { Readable } from 'node:stream'
|
|
|
|
import { nodeReadableToWebStream } from '@/lib/node-readable-stream'
|
|
|
|
async function readAll(stream: ReadableStream<Uint8Array>): Promise<Buffer> {
|
|
const reader = stream.getReader()
|
|
const chunks: Buffer[] = []
|
|
|
|
try {
|
|
while (true) {
|
|
const { value, done } = await reader.read()
|
|
if (done) {
|
|
return Buffer.concat(chunks)
|
|
}
|
|
|
|
chunks.push(Buffer.from(value))
|
|
}
|
|
} finally {
|
|
reader.releaseLock()
|
|
}
|
|
}
|
|
|
|
class CountingReadable extends Readable {
|
|
produced = 0
|
|
|
|
private nextChunk = 0
|
|
|
|
constructor(private readonly totalChunks: number) {
|
|
// One-byte high water mark keeps Node's own buffering from hiding a
|
|
// converter that eagerly drains the source.
|
|
super({ highWaterMark: 1 })
|
|
}
|
|
|
|
override _read(): void {
|
|
if (this.nextChunk >= this.totalChunks) {
|
|
this.push(null)
|
|
return
|
|
}
|
|
|
|
this.produced += 1
|
|
this.push(Buffer.from([this.nextChunk]))
|
|
this.nextChunk += 1
|
|
}
|
|
}
|
|
|
|
class CancellableReadable extends Readable {
|
|
private emitted = false
|
|
|
|
constructor() {
|
|
super({ highWaterMark: 1 })
|
|
}
|
|
|
|
override _read(): void {
|
|
// Keep the source open after the first chunk so cancellation, rather than
|
|
// natural EOF, is what must release it.
|
|
if (!this.emitted) {
|
|
this.emitted = true
|
|
this.push(Buffer.from('first chunk'))
|
|
}
|
|
}
|
|
}
|
|
|
|
describe('nodeReadableToWebStream', () => {
|
|
test('preserves every byte across the Node-to-Web conversion', async () => {
|
|
const sourceChunks = [
|
|
Buffer.from('header\0', 'utf8'),
|
|
Buffer.from([0, 1, 127, 128, 254, 255]),
|
|
Buffer.from('tail', 'utf8'),
|
|
]
|
|
const source = Readable.from(sourceChunks)
|
|
|
|
const result = await readAll(nodeReadableToWebStream(source))
|
|
|
|
expect(result).toEqual(Buffer.concat(sourceChunks))
|
|
})
|
|
|
|
test('does not pre-read every source chunk before the Web stream is consumed', async () => {
|
|
const totalChunks = 64
|
|
const source = new CountingReadable(totalChunks)
|
|
const webStream = nodeReadableToWebStream(source)
|
|
|
|
try {
|
|
// Let the stream's initial pull settle without introducing a time-based
|
|
// delay. A correctly backpressured adapter may read one chunk (and Node
|
|
// may briefly request one extra), but it must not drain the source.
|
|
await Promise.resolve()
|
|
|
|
expect(source.produced).toBeLessThan(totalChunks)
|
|
} finally {
|
|
await webStream.cancel()
|
|
}
|
|
})
|
|
|
|
test('destroys the Node stream when the Web reader is cancelled', async () => {
|
|
const source = new CancellableReadable()
|
|
const destroySpy = jest.spyOn(source, 'destroy')
|
|
const reader = nodeReadableToWebStream(source).getReader()
|
|
|
|
await expect(reader.read()).resolves.toEqual({
|
|
done: false,
|
|
value: expect.any(Uint8Array),
|
|
})
|
|
|
|
await reader.cancel()
|
|
|
|
expect(destroySpy).toHaveBeenCalledTimes(1)
|
|
expect(source.destroyed).toBe(true)
|
|
})
|
|
|
|
test('notifies completion only after the source reaches EOF', async () => {
|
|
const onComplete = jest.fn()
|
|
const source = Readable.from([Buffer.from('complete')])
|
|
|
|
const result = await readAll(nodeReadableToWebStream(source, { onComplete }))
|
|
|
|
expect(result.toString()).toBe('complete')
|
|
expect(onComplete).toHaveBeenCalledTimes(1)
|
|
})
|
|
|
|
test('does not notify completion when the reader is cancelled', async () => {
|
|
const onComplete = jest.fn()
|
|
const source = new CancellableReadable()
|
|
const reader = nodeReadableToWebStream(source, { onComplete }).getReader()
|
|
|
|
await reader.read()
|
|
await reader.cancel()
|
|
|
|
expect(onComplete).not.toHaveBeenCalled()
|
|
})
|
|
})
|