Something went wrong. Try again.
archived: streaming 7-Zip archive extractor
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103import Queue from '@mary/ds-queue';
import { TruncatedArchiveError } from '../errors.ts';import { concat } from './buffer.ts';
/** * a byte source that lazily buffers from a ReadableStream and lets callers * mix async refills with synchronous reads. matches the shape of * pkg-rar's `StreamReader`: callers `await require(size)` to ensure the * buffer can deliver at least that many bytes, then `take(size)` to pull * them out. */export class StreamReader implements AsyncDisposable { #reader: ReadableStreamDefaultReader<Uint8Array>;
#chunks = new Queue<Uint8Array>(); #buffered = 0; #consumed = 0; #done = false;
constructor(stream: ReadableStream<Uint8Array>) { this.#reader = stream.getReader(); }
/** bytes available without an await. */ get buffered(): number { return this.#buffered; }
/** total bytes pulled from this reader so far. */ get consumed(): number { return this.#consumed; }
/** whether the upstream has signalled end-of-stream. */ get done(): boolean { return this.#done; }
/** * refills the internal buffer until at least `size` bytes are available. * does not consume them; call {@link take} after a successful require. * * @throws {@link TruncatedArchiveError} when the stream ends before * delivering enough bytes */ async require(size: number): Promise<void> { while (this.#buffered < size) { const { done, value } = await this.#reader.read(); if (done) { this.#done = true; throw new TruncatedArchiveError( `unexpected end of stream: expected ${size} bytes, got ${this.#buffered}`, ); } this.#chunks.enqueue(value); this.#buffered += value.length; } }
/** * pulls exactly `size` bytes from the buffer. fast path when the size * fits inside a single internal chunk; otherwise concatenates. */ 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) { 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]; } return concat(needed, size); }
async [Symbol.asyncDispose](): Promise<void> { await this.#reader.cancel(); }}