| 606 | |
| 607 | // Import compression transform |
| 608 | function createGzipTransform(): TransformObject { |
| 609 | const gzip = zlib.createGzip(); |
| 610 | const pending: Uint8Array[] = []; |
| 611 | let error: Error | null = null; |
| 612 | |
| 613 | gzip.on('data', (chunk: Buffer) => pending.push(new Uint8Array(chunk))); |
| 614 | gzip.on('error', (err) => { error = err; }); |
| 615 | |
| 616 | async function processChunk(chunk: Uint8Array | null): Promise<Uint8Array[]> { |
| 617 | if (error) throw error; |
| 618 | if (chunk === null) { |
| 619 | await new Promise<void>((resolve, reject) => { |
| 620 | gzip.once('end', resolve); |
| 621 | gzip.once('error', reject); |
| 622 | gzip.end(); |
| 623 | }); |
| 624 | return pending.splice(0); |
| 625 | } |
| 626 | await new Promise<void>((resolve, reject) => { |
| 627 | gzip.write(copyToBuffer(chunk), (err) => { |
| 628 | if (err) { reject(err); return; } |
| 629 | gzip.flush(() => resolve()); |
| 630 | }); |
| 631 | }); |
| 632 | return pending.splice(0); |
| 633 | } |
| 634 | |
| 635 | return { |
| 636 | |
| 637 | async *transform(source: AsyncIterable<Uint8Array[] | null>, { signal }: { signal: AbortSignal }) { |
| 638 | const onAbort = () => { |
| 639 | gzip.destroy(signal.reason instanceof Error ? signal.reason : new Error(String(signal.reason))); |
| 640 | }; |
| 641 | signal.addEventListener('abort', onAbort, { once: true }); |
| 642 | try { |
| 643 | for await (const batches of source) { |
| 644 | if (batches === null) { |
| 645 | const output = await processChunk(null); |
| 646 | for (const chunk of output) yield chunk; |
| 647 | continue; |
| 648 | } |
| 649 | for (const chunk of batches) { |
| 650 | const output = await processChunk(chunk); |
| 651 | for (const out of output) yield out; |
| 652 | } |
| 653 | } |
| 654 | } finally { |
| 655 | signal.removeEventListener('abort', onAbort); |
| 656 | gzip.destroy(); |
| 657 | } |
| 658 | }, |
| 659 | }; |
| 660 | } |
| 661 | |
| 662 | function createGunzipTransform(): TransformObject { |
| 663 | const gunzip = zlib.createGunzip(); |