mary.my.id
streaming 7-Zip archive extractor
1/**
2 * @module
3 *
4 * a {@link Reader} that concatenates a sequence of underlying readers into
5 * one logical byte source. used to span multi-volume 7z archives — the
6 * signature header lives in volume 1, the data area and NextHeader span
7 * the whole set.
8 */
9
10import type { Reader } from './types.ts';
11
12/**
13 * combines an ordered list of readers into a single virtual reader; reads that
14 * cross a volume boundary span the underlying readers in order.
15 *
16 * @param readers ordered readers; the first holds the signature header, the
17 * last holds the NextHeader
18 * @returns a reader exposing the concatenated byte sequence
19 */
20export function concatReaders(readers: readonly Reader[]): Reader {
21 if (readers.length === 0) {
22 throw new Error('concatReaders requires at least one reader');
23 }
24 if (readers.length === 1) {
25 return readers[0];
26 }
27
28 const offsets: number[] = [];
29 let total = 0;
30 for (const r of readers) {
31 offsets.push(total);
32 total += r.length;
33 }
34
35 const findVolume = (absoluteOffset: number): { idx: number; localOffset: number } => {
36 // linear search is fine — multi-volume archives have typically <10 parts.
37 for (let i = readers.length - 1; i >= 0; i--) {
38 if (offsets[i] <= absoluteOffset) {
39 return { idx: i, localOffset: absoluteOffset - offsets[i] };
40 }
41 }
42 return { idx: 0, localOffset: absoluteOffset };
43 };
44
45 return {
46 length: total,
47 // deno-lint-ignore require-await
48 async read(offset: number, length?: number): Promise<ReadableStream<Uint8Array<ArrayBuffer>>> {
49 const end = length !== undefined ? offset + length : total;
50 let pos = offset;
51
52 const segments: { reader: Reader; localOffset: number; localLength: number }[] = [];
53 while (pos < end) {
54 const { idx, localOffset } = findVolume(pos);
55 const reader = readers[idx];
56 const available = reader.length - localOffset;
57 const want = Math.min(end - pos, available);
58 if (want === 0) {
59 break;
60 }
61 segments.push({ localLength: want, localOffset, reader });
62 pos += want;
63 }
64
65 let segIdx = 0;
66 let inner: ReadableStreamDefaultReader<Uint8Array<ArrayBuffer>> | undefined;
67 let cancelled = false;
68
69 return new ReadableStream({
70 async pull(controller) {
71 while (true) {
72 if (cancelled) {
73 controller.close();
74 return;
75 }
76 if (inner === undefined) {
77 if (segIdx >= segments.length) {
78 controller.close();
79 return;
80 }
81 const seg = segments[segIdx++];
82 const stream = await seg.reader.read(seg.localOffset, seg.localLength);
83 if (cancelled) {
84 await stream.cancel();
85 controller.close();
86 return;
87 }
88 inner = stream.getReader();
89 }
90 const { done, value } = await inner.read();
91 if (done) {
92 inner = undefined;
93 continue;
94 }
95 controller.enqueue(value);
96 return;
97 }
98 },
99 async cancel(reason) {
100 cancelled = true;
101 await inner?.cancel(reason);
102 },
103 });
104 },
105 };
106}