Something went wrong. Try again.
archived: streaming zip archiver/extractor jsr.io/@mary/zip
typescript jsr
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495import Queue from '@mary/ds-queue';
import { TruncatedArchiveError } from '../errors.ts';import { concat } from './buffer.ts';
export class StreamReader implements AsyncDisposable { #reader: ReadableStreamDefaultReader<Uint8Array>;
#chunks = new Queue<Uint8Array>(); #buffered = 0; #consumed = 0;
constructor(stream: ReadableStream<Uint8Array>) { this.#reader = stream.getReader(); }
get buffered(): number { return this.#buffered; }
get consumed(): number { return this.#consumed; }
async require(size: number): Promise<void> { const reader = this.#reader; const chunks = this.#chunks;
while (this.#buffered < size) { const { done, value } = await reader.read();
if (done) { throw new TruncatedArchiveError( `unexpected end of stream: expected ${size} bytes, got ${this.#buffered}`, ); }
chunks.enqueue(value);
this.#buffered += value.length; } }
take(size: number): Uint8Array { if (size === 0) { return new Uint8Array(0); }
if (this.#buffered < size) { throw new TruncatedArchiveError( `unexpected end of stream: expected ${size} bytes, got ${this.#buffered}`, ); }
const chunks = this.#chunks; const needed: Uint8Array[] = [];
let taken = 0;
while (taken < size) { const chunk = chunks.dequeue()!; const chunkLength = chunk.length;
const took = Math.min(chunkLength, size - taken);
if (took < chunkLength) { // Split chunk needed.push(chunk.subarray(0, took));
chunks.enqueueFront(chunk.subarray(took));
taken += took; break; } else { needed.push(chunk);
taken += chunkLength; } }
this.#buffered -= size; this.#consumed += size;
if (needed.length === 1) { return needed[0]; } else { return concat(needed, size); } }
async [Symbol.asyncDispose](): Promise<void> { await this.#reader.cancel(); }}