Feature: ZipReader and ZipWriter as Streams via .pipeThrough

[BlackAsLight]

Hello, while this project does offer the services I am in need of, it doesn't quite offer it in the way I'd like to consume it. Specifically when dealing with streams. I very much like to stream in zip files from external services, unzip them in memory, modify the contents and rezip them to save, all without having to write any step, but the final if I so choose, to the disk.

I did come up with two working classes to wrap the existing ZipReader and ZipWriter classes that are instead TransformStreams so one can place them inside a readable.pipeThrough and I've found that they work quite well. I do hope that you'd consider adding these classes or something similar to achieve the same effect.

Wrapper for ZipReader

import { Entry, ZipReader, ZipReaderConstructorOptions } from 'https://deno.land/x/[email protected]/index.js'

export class ZipDecompressStream<T> {
	readable: ReadableStream<Omit<Entry, 'getData'> & { readable?: ReadableStream<Uint8Array> }>
	writable: WritableStream<T>
	constructor (options?: ZipReaderConstructorOptions) {
		const { readable, writable } = new TransformStream<T, T>()
		const gen = new ZipReader(readable, options).getEntriesGenerator()
		this.readable = new ReadableStream({
			async pull(controller) {
				const { done, value } = await gen.next()
				if (done)
					return controller.close()
				const chunk = {
					...value,
					readable: (function () {
						const { readable, writable } = new TransformStream<Uint8Array, Uint8Array>()
						if (value.getData) {
							value.getData(writable)
							return readable
						}
					})()
				}
				delete chunk.getData
				controller.enqueue(chunk)
			}
		})
		this.writable = writable
	}
}

Wrapper for ZipWriter

import { ZipWriter, ZipWriterConstructorOptions } from 'https://deno.land/x/[email protected]/index.js'

export class ZipCompressStream<T> {
	readable: ReadableStream<Uint8Array>
	writable: WritableStream<T>
	constructor (path: string, options?: ZipWriterConstructorOptions) {
		const { readable, writable } = new TransformStream<T, Uint8Array>()
		const zipWriter = new ZipWriter(writable, options) // Out-Bound Compressed
		zipWriter.add(path, (() => {
			const { readable, writable } = new TransformStream<T, T>({
				flush() {
					zipWriter.close()
				}
			})
			this.writable = writable // In-Bound Uncompressed
			return readable // In-Bound Uncompressed
		})())
		this.readable = readable // Out-Bound Uncompressed
	}
}

Example

This working example just streams in a zip file from the web, decompresses it and then recompresses it before saving it to disk.

// Required Setup
import { DOMParser } from 'https://deno.land/x/[email protected]/deno-dom-wasm.ts'

const url = 'https://politicsandwar.com/data/trades/trades-'
	+ new DOMParser()
		.parseFromString(await (await fetch('https://politicsandwar.com/data/trades/?C=M;O=A')).text(), 'text/html')!
		.querySelector('a[href*="trade"]')!
		.getAttribute('href')!
		.slice(7, -8)
	+ '.csv.zip'

// Working Example
for await (const entry of (await fetch(url)).body!.pipeThrough(new ZipDecompressStream()))
	if (entry.readable) 
		entry.readable
			.pipeThrough(new ZipCompressStream(entry.filename))
			.pipeTo((await Deno.create(entry.filename + '.zip')).writable)

It should be noted that with ZipDecompressStream, if the developer doesn't completely consume the readable stream or call .cancel() (or use a method that would call .cancel() like a for await of) on it then it will hang around in memory forever, even if it goes out of scope.

import { CsvParseStream } from 'https://jsr.io/@std/csv/0.216.0/csv_parse_stream.ts'

for await ( const entry of (await fetch(url)).body!.pipeThrough(new ZipDecompressStream()))
	if (entry.readable) {
		const readable = entry.readable
			.pipeThrough(new TextDecoderStream())
			.pipeThrough(new CsvParseStream())
		console.log(await readable[ Symbol.asyncIterator ]().next())
	}