wehub-resource-sync
101 lines
3.8 KiB
TypeScript
101 lines
3.8 KiB
TypeScript
/**
|
|
* Covers the min-chunk coalescing Transform used between the aws-chunked
|
|
* parsers and the AWS SDK client. The parsers emit whatever TCP fragment
|
|
* arrives, which is often <8 KiB — real S3 / MinIO reject non-final
|
|
* streaming chunks that small. The coalescer guarantees ≥minBytes writes
|
|
* with one smaller flush at end-of-stream.
|
|
*
|
|
* It lives inline in put-object.ts and upload-part.ts so we don't export
|
|
* it from a shared module; the test below inlines a copy to lock the
|
|
* contract down.
|
|
*/
|
|
import { describe, it, expect } from 'vitest';
|
|
import { Readable, Transform, TransformCallback } from 'stream';
|
|
|
|
class MinChunkSizeStream extends Transform {
|
|
private pending: Buffer[] = [];
|
|
private pendingLen = 0;
|
|
constructor(private readonly minBytes: number) {
|
|
super();
|
|
}
|
|
_transform(chunk: Buffer, _enc: BufferEncoding, cb: TransformCallback): void {
|
|
this.pending.push(chunk);
|
|
this.pendingLen += chunk.length;
|
|
if (this.pendingLen >= this.minBytes) {
|
|
this.push(Buffer.concat(this.pending, this.pendingLen));
|
|
this.pending = [];
|
|
this.pendingLen = 0;
|
|
}
|
|
cb();
|
|
}
|
|
_flush(cb: TransformCallback): void {
|
|
if (this.pendingLen > 0) {
|
|
this.push(Buffer.concat(this.pending, this.pendingLen));
|
|
this.pending = [];
|
|
this.pendingLen = 0;
|
|
}
|
|
cb();
|
|
}
|
|
}
|
|
|
|
async function pipeAndCollect(
|
|
inputs: Buffer[],
|
|
minBytes: number
|
|
): Promise<{ chunks: Buffer[]; combined: Buffer }> {
|
|
const stream = new MinChunkSizeStream(minBytes);
|
|
Readable.from(inputs).pipe(stream);
|
|
const chunks: Buffer[] = [];
|
|
for await (const c of stream) chunks.push(c as Buffer);
|
|
return { chunks, combined: Buffer.concat(chunks) };
|
|
}
|
|
|
|
describe('MinChunkSizeStream', () => {
|
|
it('coalesces small writes into ≥minBytes emissions', async () => {
|
|
const small = Array.from({ length: 100 }, () => Buffer.alloc(100, 0x61));
|
|
const { chunks, combined } = await pipeAndCollect(small, 8192);
|
|
|
|
// All emissions before the last must be ≥minBytes; only the final
|
|
// flush is allowed to be smaller.
|
|
for (let i = 0; i < chunks.length - 1; i++) {
|
|
expect(chunks[i].length).toBeGreaterThanOrEqual(8192);
|
|
}
|
|
expect(combined.length).toBe(10000);
|
|
});
|
|
|
|
it('passes a single large buffer through untouched', async () => {
|
|
const big = Buffer.alloc(100_000, 0x7a);
|
|
const { chunks, combined } = await pipeAndCollect([big], 65536);
|
|
expect(chunks).toHaveLength(1);
|
|
expect(combined.equals(big)).toBe(true);
|
|
});
|
|
|
|
it('flushes any remaining bytes on stream end (final chunk may be <minBytes)', async () => {
|
|
const inputs = [Buffer.alloc(5000, 0x31), Buffer.alloc(5000, 0x32)]; // 10000 < 65536
|
|
const { chunks, combined } = await pipeAndCollect(inputs, 65536);
|
|
expect(chunks).toHaveLength(1);
|
|
expect(chunks[0].length).toBe(10000);
|
|
expect(combined[0]).toBe(0x31);
|
|
expect(combined[combined.length - 1]).toBe(0x32);
|
|
});
|
|
|
|
it('preserves byte order across coalesced emissions', async () => {
|
|
// 4 writes of 25_000 bytes each = 100 KiB; coalesced at 65 KiB threshold
|
|
const a = Buffer.alloc(25_000, 0xa);
|
|
const b = Buffer.alloc(25_000, 0xb);
|
|
const c = Buffer.alloc(25_000, 0xc);
|
|
const d = Buffer.alloc(25_000, 0xd);
|
|
const { combined } = await pipeAndCollect([a, b, c, d], 65536);
|
|
expect(combined.length).toBe(100_000);
|
|
expect(combined.slice(0, 25_000).every((v) => v === 0xa)).toBe(true);
|
|
expect(combined.slice(25_000, 50_000).every((v) => v === 0xb)).toBe(true);
|
|
expect(combined.slice(50_000, 75_000).every((v) => v === 0xc)).toBe(true);
|
|
expect(combined.slice(75_000).every((v) => v === 0xd)).toBe(true);
|
|
});
|
|
|
|
it('emits nothing when input is empty', async () => {
|
|
const { chunks, combined } = await pipeAndCollect([], 8192);
|
|
expect(chunks).toHaveLength(0);
|
|
expect(combined.length).toBe(0);
|
|
});
|
|
});
|