Skip to content

Commit f92ef4e

Browse files
pimterrypimterry
committed
Add tests to confirm real streaming is happening
1 parent 9a72c48 commit f92ef4e

1 file changed

Lines changed: 265 additions & 32 deletions

File tree

‎test/stream.spec.ts‎

Lines changed: 265 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -20,38 +20,6 @@ import {
2020
zstdDecompress
2121
} from '../src/index';
2222

23-
// Helper to collect all chunks from a ReadableStream into a single Uint8Array
24-
async function collectStream(stream: ReadableStream<Uint8Array>): Promise<Uint8Array> {
25-
const reader = stream.getReader();
26-
const chunks: Uint8Array[] = [];
27-
let totalLength = 0;
28-
29-
while (true) {
30-
const { done, value } = await reader.read();
31-
if (done) break;
32-
chunks.push(value);
33-
totalLength += value.length;
34-
}
35-
36-
const result = new Uint8Array(totalLength);
37-
let offset = 0;
38-
for (const chunk of chunks) {
39-
result.set(chunk, offset);
40-
offset += chunk.length;
41-
}
42-
return result;
43-
}
44-
45-
// Helper to create a ReadableStream from a Uint8Array
46-
function createReadableStream(data: Uint8Array): ReadableStream<Uint8Array> {
47-
return new ReadableStream({
48-
start(controller) {
49-
controller.enqueue(data);
50-
controller.close();
51-
}
52-
});
53-
}
54-
5523
describe("Streaming", () => {
5624
describe("Gzip", () => {
5725
it('should compress data with gzip stream', async () => {
@@ -132,6 +100,39 @@ describe("Streaming", () => {
132100
const decompressed = zlib.gunzipSync(Buffer.from(compressed));
133101
expect(decompressed.toString()).to.equal(repeated);
134102
});
103+
104+
it('should produce incremental output during compression', async () => {
105+
const inputChunks = generateLargeData(STREAMING_TEST_SIZE_KB);
106+
const { outputBeforeEnd, totalOutput } = await testTrueStreaming(
107+
createGzipStream(),
108+
inputChunks
109+
);
110+
111+
expect(outputBeforeEnd).to.equal(true, 'Expected output before input completed');
112+
113+
const decompressed = zlib.gunzipSync(Buffer.from(totalOutput));
114+
const expectedSize = inputChunks.reduce((sum, c) => sum + c.length, 0);
115+
expect(decompressed.length).to.equal(expectedSize);
116+
});
117+
118+
it('should produce incremental output during decompression', async () => {
119+
const originalData = Buffer.concat(generateLargeData(STREAMING_TEST_SIZE_KB));
120+
const compressed = zlib.gzipSync(originalData);
121+
122+
const chunkSize = 16 * 1024;
123+
const inputChunks: Uint8Array[] = [];
124+
for (let i = 0; i < compressed.length; i += chunkSize) {
125+
inputChunks.push(compressed.subarray(i, Math.min(i + chunkSize, compressed.length)));
126+
}
127+
128+
const { outputBeforeEnd, totalOutput } = await testTrueStreaming(
129+
createGunzipStream(),
130+
inputChunks
131+
);
132+
133+
expect(outputBeforeEnd).to.equal(true, 'Expected output before input completed');
134+
expect(totalOutput.length).to.equal(originalData.length);
135+
});
135136
});
136137

137138
describe("Deflate", () => {
@@ -168,6 +169,39 @@ describe("Streaming", () => {
168169

169170
expect(Buffer.from(result).toString()).to.equal(original);
170171
});
172+
173+
it('should produce incremental output during compression', async () => {
174+
const inputChunks = generateLargeData(STREAMING_TEST_SIZE_KB);
175+
const { outputBeforeEnd, totalOutput } = await testTrueStreaming(
176+
createDeflateStream(),
177+
inputChunks
178+
);
179+
180+
expect(outputBeforeEnd).to.equal(true, 'Expected output before input completed');
181+
182+
const decompressed = zlib.inflateSync(Buffer.from(totalOutput));
183+
const expectedSize = inputChunks.reduce((sum, c) => sum + c.length, 0);
184+
expect(decompressed.length).to.equal(expectedSize);
185+
});
186+
187+
it('should produce incremental output during decompression', async () => {
188+
const originalData = Buffer.concat(generateLargeData(STREAMING_TEST_SIZE_KB));
189+
const compressed = zlib.deflateSync(originalData);
190+
191+
const chunkSize = 16 * 1024;
192+
const inputChunks: Uint8Array[] = [];
193+
for (let i = 0; i < compressed.length; i += chunkSize) {
194+
inputChunks.push(compressed.subarray(i, Math.min(i + chunkSize, compressed.length)));
195+
}
196+
197+
const { outputBeforeEnd, totalOutput } = await testTrueStreaming(
198+
createInflateStream(),
199+
inputChunks
200+
);
201+
202+
expect(outputBeforeEnd).to.equal(true, 'Expected output before input completed');
203+
expect(totalOutput.length).to.equal(originalData.length);
204+
});
171205
});
172206

173207
describe("Deflate-Raw", () => {
@@ -256,6 +290,49 @@ describe("Streaming", () => {
256290
const decompressed = await brotliDecompress(compressed);
257291
expect(Buffer.from(decompressed).toString()).to.equal(repeated);
258292
});
293+
294+
// Brotli has a larger internal buffer than gzip/deflate, requiring ~1.3MB of
295+
// incompressible data before it produces streaming output. With compressible data
296+
// the output is too small to trigger streaming. We use 2MB of random data here.
297+
it('should produce incremental output during compression', async function () {
298+
this.timeout(10000); // Can be slow
299+
300+
const BROTLI_DATA_SIZE_KB = 2048;
301+
const inputChunks = generateLargeData(BROTLI_DATA_SIZE_KB, false);
302+
const { outputBeforeEnd, totalOutput } = await testTrueStreaming(
303+
createBrotliCompressStream(),
304+
inputChunks
305+
);
306+
307+
expect(outputBeforeEnd).to.equal(true, 'Expected output before input completed');
308+
309+
const decompressed = await brotliDecompress(totalOutput);
310+
const expectedSize = inputChunks.reduce((sum, c) => sum + c.length, 0);
311+
expect(decompressed.length).to.equal(expectedSize);
312+
});
313+
314+
it('should produce incremental output during decompression', async function () {
315+
this.timeout(10000); // Can be slow
316+
317+
const BROTLI_DECOMPRESS_SIZE_KB = 1024; // 1MB
318+
const originalData = Buffer.concat(generateLargeData(BROTLI_DECOMPRESS_SIZE_KB, false));
319+
const compressed = await brotliCompress(originalData);
320+
321+
const chunkSize = 16 * 1024;
322+
const inputChunks: Uint8Array[] = [];
323+
for (let i = 0; i < compressed.length; i += chunkSize) {
324+
inputChunks.push(new Uint8Array(compressed.buffer, compressed.byteOffset + i,
325+
Math.min(chunkSize, compressed.length - i)));
326+
}
327+
328+
const { outputBeforeEnd, totalOutput } = await testTrueStreaming(
329+
createBrotliDecompressStream(),
330+
inputChunks
331+
);
332+
333+
expect(outputBeforeEnd).to.equal(true, 'Expected output before input completed');
334+
expect(totalOutput.length).to.equal(originalData.length);
335+
});
259336
});
260337

261338
describe("Zstd", () => {
@@ -308,5 +385,161 @@ describe("Streaming", () => {
308385
const decompressed = await zstdDecompress(compressed);
309386
expect(Buffer.from(decompressed).toString()).to.equal(repeated);
310387
});
388+
389+
it('should produce incremental output during compression', async () => {
390+
const inputChunks = generateLargeData(STREAMING_TEST_SIZE_KB);
391+
const { outputBeforeEnd, totalOutput } = await testTrueStreaming(
392+
createZstdCompressStream(),
393+
inputChunks
394+
);
395+
396+
expect(outputBeforeEnd).to.equal(true, 'Expected output before input completed');
397+
398+
const decompressed = await zstdDecompress(totalOutput);
399+
const expectedSize = inputChunks.reduce((sum, c) => sum + c.length, 0);
400+
expect(decompressed.length).to.equal(expectedSize);
401+
});
402+
403+
it('should produce incremental output during decompression', async () => {
404+
const originalData = Buffer.concat(generateLargeData(STREAMING_TEST_SIZE_KB));
405+
const compressed = await zstdCompress(originalData);
406+
407+
const chunkSize = 16 * 1024;
408+
const inputChunks: Uint8Array[] = [];
409+
for (let i = 0; i < compressed.length; i += chunkSize) {
410+
inputChunks.push(new Uint8Array(compressed.buffer, compressed.byteOffset + i,
411+
Math.min(chunkSize, compressed.length - i)));
412+
}
413+
414+
const { outputBeforeEnd, totalOutput } = await testTrueStreaming(
415+
createZstdDecompressStream(),
416+
inputChunks
417+
);
418+
419+
expect(outputBeforeEnd).to.equal(true, 'Expected output before input completed');
420+
expect(totalOutput.length).to.equal(originalData.length);
421+
});
311422
});
312423
});
424+
425+
// Helper to collect all chunks from a ReadableStream into a single Uint8Array
426+
async function collectStream(stream: ReadableStream<Uint8Array>): Promise<Uint8Array> {
427+
const reader = stream.getReader();
428+
const chunks: Uint8Array[] = [];
429+
let totalLength = 0;
430+
431+
while (true) {
432+
const { done, value } = await reader.read();
433+
if (done) break;
434+
chunks.push(value);
435+
totalLength += value.length;
436+
}
437+
438+
const result = new Uint8Array(totalLength);
439+
let offset = 0;
440+
for (const chunk of chunks) {
441+
result.set(chunk, offset);
442+
offset += chunk.length;
443+
}
444+
return result;
445+
}
446+
447+
// Helper to create a ReadableStream from a Uint8Array
448+
function createReadableStream(data: Uint8Array): ReadableStream<Uint8Array> {
449+
return new ReadableStream({
450+
start(controller) {
451+
controller.enqueue(data);
452+
controller.close();
453+
}
454+
});
455+
}
456+
457+
// Data size for incremental output tests (256KB is enough for gzip/deflate/zstd)
458+
const STREAMING_TEST_SIZE_KB = 256;
459+
460+
// Helper to test true streaming behavior - verifies output arrives before all input is sent
461+
async function testTrueStreaming(
462+
transformStream: TransformStream<Uint8Array, Uint8Array>,
463+
inputChunks: Uint8Array[]
464+
): Promise<{ outputBeforeEnd: boolean; totalOutput: Uint8Array }> {
465+
const outputChunks: Uint8Array[] = [];
466+
let outputReceivedBeforeEnd = false;
467+
let inputComplete = false;
468+
469+
const writer = transformStream.writable.getWriter();
470+
const reader = transformStream.readable.getReader();
471+
472+
// Start reading in background - this allows output to be collected as it arrives
473+
const readPromise = (async () => {
474+
while (true) {
475+
const { done, value } = await reader.read();
476+
if (done) break;
477+
outputChunks.push(value);
478+
if (!inputComplete) {
479+
outputReceivedBeforeEnd = true;
480+
}
481+
}
482+
})();
483+
484+
// Write chunks with small delays to allow output processing
485+
for (const chunk of inputChunks) {
486+
await writer.write(chunk);
487+
await new Promise(resolve => setTimeout(resolve, 0));
488+
}
489+
490+
inputComplete = true;
491+
await writer.close();
492+
await readPromise;
493+
494+
// Combine output chunks
495+
const totalLength = outputChunks.reduce((sum, chunk) => sum + chunk.length, 0);
496+
const totalOutput = new Uint8Array(totalLength);
497+
let offset = 0;
498+
for (const chunk of outputChunks) {
499+
totalOutput.set(chunk, offset);
500+
offset += chunk.length;
501+
}
502+
503+
return { outputBeforeEnd: outputReceivedBeforeEnd, totalOutput };
504+
}
505+
506+
// Mulberry32: high-quality 32-bit seeded PRNG that produces incompressible output
507+
function createSeededRng(seed: number) {
508+
return function(): number {
509+
let t = seed += 0x6D2B79F5;
510+
t = Math.imul(t ^ t >>> 15, t | 1);
511+
t ^= t + Math.imul(t ^ t >>> 7, t | 61);
512+
return ((t ^ t >>> 14) >>> 0) / 4294967296;
513+
};
514+
}
515+
516+
// Generate large test data that will produce streaming output
517+
function generateLargeData(sizeKB: number, compressible = true): Uint8Array[] {
518+
const chunkSize = 16 * 1024; // 16KB chunks
519+
const totalSize = sizeKB * 1024;
520+
const chunks: Uint8Array[] = [];
521+
const pattern = 'This is test data that will be repeated to create compressible content. ';
522+
523+
// Fixed seed for deterministic, reproducible test data
524+
const rng = createSeededRng(0x12345678);
525+
526+
let remaining = totalSize;
527+
while (remaining > 0) {
528+
const size = Math.min(chunkSize, remaining);
529+
const chunk = new Uint8Array(size);
530+
if (compressible) {
531+
for (let i = 0; i < size; i++) {
532+
chunk[i] = pattern.charCodeAt(i % pattern.length);
533+
}
534+
} else {
535+
// Use seeded PRNG for deterministic incompressible data
536+
for (let i = 0; i < size; i++) {
537+
chunk[i] = Math.floor(rng() * 256);
538+
}
539+
}
540+
chunks.push(chunk);
541+
remaining -= size;
542+
}
543+
544+
return chunks;
545+
}

0 commit comments

Comments
 (0)