1/**
2 * this module contains helpers for creating and consuming `ReadableStream`,
3 * particularly with binary payloads.
4 *
5 * @module
6 */
7
8const shared = new Uint8Array(8);
9const sharedView = new DataView(shared.buffer);
10
11/** write bytes `data` to a destination */
12export type WriteBytesFn = (data: Uint8Array) => void;
13
14/** something that can be written to */
15export type WriteTarget = WriteBytesFn | { enqueue(chunk: Uint8Array): void; close?(): void };
16
17/**
18 * `ReadableStreamDefaultController.enqueue` does not buffer data in any way.
19 * for use cases such as a streaming write of many small binary chunks, such as
20 * differently sized integers in the case of `progress.encodeByteStream`, this
21 * batches those many small writes into their own `ArrayBuffer` objects.
22 *
23 * when passing a `ReadableStreamDefaultController`, the `BufferedWriter`
24 * "owns" it, and will close the stream when disposed. to prevent that, pass
25 * a `WriteBytesFn` instead.
26 *
27 * ```ts
28 * const w = new BufferedWriter(unbufferedWrite);
29 * w.write(string.encodeUtf8("a null-terminated string"));
30 * w.u8(0);
31 * w.write(string.encodeUtf8("a second null-terminated string"));
32 * w.u8(0);
33 *
34 * w.flush(); // don't forget to flush!
35 * // `unbufferedWrite` will be called once with 57 bytes
36 * ```
37 */
38export class BufferedWriter {
39 #view: DataView<ArrayBuffer>;
40 #written: number = 0;
41 #flush: WriteBytesFn;
42 #size: number;
43 #onClose?: (() => void) | null | undefined;
44
45 constructor(flush: WriteTarget, size: number = 8192) {
46 this.#flush = typeof flush === "function" ? flush : flush.enqueue.bind(flush);
47 this.#view = new DataView(new ArrayBuffer(size));
48 this.#size = size;
49 this.#onClose = typeof flush === "function" ? null : flush.close?.bind(flush);
50 }
51
52 flush() {
53 const written = this.#written;
54 if (written === 0) return;
55 const size = this.#size;
56 if (written === size) {
57 // transfer
58 this.#flush(new Uint8Array(this.#view.buffer));
59 } else {
60 // copy
61 this.#flush(new Uint8Array(this.#view.buffer.slice(0, written)));
62 }
63 this.#view = new DataView(new ArrayBuffer(size));
64 this.#written = 0;
65 }
66
67 close() {
68 const written = this.#written;
69 if (written === 0) return;
70 const size = this.#size;
71 if (written === size) {
72 // transfer
73 this.#flush(new Uint8Array(this.#view.buffer));
74 } else {
75 // copy
76 this.#flush(new Uint8Array(this.#view.buffer.slice(0, written)));
77 }
78 this.#written = 0;
79 this.#size = 0;
80 this.#view = new DataView(new ArrayBuffer(0));
81 this.#onClose?.();
82 }
83
84 [Symbol.dispose]() {
85 this.close();
86 }
87
88 write(bytes: Uint8Array<ArrayBuffer> | ArrayBuffer) {
89 const len = bytes.byteLength;
90 if (len >= this.size) {
91 this.flush();
92 this.#flush("buffer" in bytes ? bytes : new Uint8Array(bytes));
93 return;
94 }
95 this.ensureUnusedCapacity(len);
96 new Uint8Array(this.#view.buffer, this.#written).set(
97 bytes instanceof ArrayBuffer ? new Uint8Array(bytes) : bytes,
98 );
99 this.#written += bytes.byteLength;
100 }
101
102 varUint(int: number) {
103 ASSERT(int >= 0);
104 do {
105 const data = int % (1 << 7);
106 int = Math.floor(int / (1 << 7));
107 const continuation = int > 0 ? 1 << 7 : 0;
108 this.u8(data + continuation);
109 } while (int > 0);
110 }
111
112 u8(int: number) {
113 const size = 1;
114 this.ensureUnusedCapacity(size);
115 this.#view.setUint8(this.#written, int);
116 this.#written += size;
117 }
118
119 u16(int: number) {
120 const size = 2;
121 this.ensureUnusedCapacity(size);
122 this.#view.setUint16(this.#written, int, true);
123 this.#written += size;
124 }
125
126 u32(int: number) {
127 const size = 4;
128 this.ensureUnusedCapacity(size);
129 this.#view.setUint32(this.#written, int, true);
130 this.#written += size;
131 }
132
133 u64(int: bigint) {
134 const size = 8;
135 this.ensureUnusedCapacity(size);
136 this.#view.setBigUint64(this.#written, int, true);
137 this.#written += size;
138 }
139
140 i8(int: number) {
141 const size = 1;
142 this.ensureUnusedCapacity(size);
143 this.#view.setUint8(this.#written, int);
144 this.#written += size;
145 }
146
147 i16(int: number) {
148 const size = 2;
149 this.ensureUnusedCapacity(size);
150 this.#view.setInt16(this.#written, int, true);
151 this.#written += size;
152 }
153
154 i32(int: number) {
155 const size = 4;
156 this.ensureUnusedCapacity(size);
157 this.#view.setInt32(this.#written, int, true);
158 this.#written += size;
159 }
160
161 i64(int: bigint) {
162 const size = 8;
163 this.ensureUnusedCapacity(size);
164 this.#view.setBigInt64(this.#written, int, true);
165 this.#written += size;
166 }
167
168 f16(float: number) {
169 const size = 2;
170 this.ensureUnusedCapacity(size);
171 this.#view.setFloat16(this.#written, float, true);
172 this.#written += size;
173 }
174
175 f32(float: number) {
176 const size = 4;
177 this.ensureUnusedCapacity(size);
178 this.#view.setFloat32(this.#written, float, true);
179 this.#written += size;
180 }
181
182 f64(float: number) {
183 const size = 8;
184 this.ensureUnusedCapacity(size);
185 this.#view.setFloat64(this.#written, float, true);
186 this.#written += size;
187 }
188
189 stringWithLength(text: string) {
190 const buf = string.encodeUtf8(text);
191 this.varUint(buf.byteLength);
192 this.write(buf);
193 }
194
195 get written(): number {
196 return this.#written;
197 }
198 get size(): number {
199 return this.#size;
200 }
201
202 ensureUnusedCapacity(extra: number) {
203 const size = this.size;
204 const written = this.#written;
205 if (written + extra > size) {
206 if (extra > size) {
207 this.resize(extra);
208 } else {
209 this.flush();
210 }
211 }
212 }
213
214 resize(newCapacity: number) {
215 newCapacity = Math.round(newCapacity);
216 const written = this.#written;
217 if (written > 0) {
218 // transfer
219 this.#flush(new Uint8Array(this.#view.buffer, 0, written));
220 this.#written = 0;
221 }
222 this.#size = newCapacity;
223 this.#view = new DataView(new ArrayBuffer(newCapacity));
224 }
225}
226
227/**
228 * a reader for `ReadableStream` which allows the caller to specify the
229 * chunking size and read numbers at a time.
230 */
231export class BufferedReader implements Disposable {
232 #reader: ReadableStreamDefaultReader<Uint8Array> | null;
233 #offset = 0;
234 #buffer: Uint8Array[] = [];
235 #available = 0;
236
237 constructor(reader: ReadableStreamDefaultReader<Uint8Array>) {
238 this.#reader = reader;
239 }
240
241 get available(): number {
242 return this.#available;
243 }
244
245 /** Releasing will discard the bytes remaining in the buffer. */
246 releaseLock() {
247 UNWRAP(this.#reader, "reader disposed").releaseLock();
248 this.#reader = null;
249 }
250
251 cancel(error: unknown) {
252 UNWRAP(this.#reader, "reader disposed").cancel(error);
253 }
254
255 async ensureAvailableOrFalse(size: number): Promise<boolean> {
256 if (size <= 0 || this.#available >= size) return true;
257 const reader = UNWRAP(this.#reader, "reader disposed");
258 const buffer = this.#buffer;
259 while (this.#available < size) {
260 const { value, done } = await reader.read();
261 if (done) break;
262 buffer.push(value);
263 this.#available += value.byteLength;
264 }
265 return this.#available >= size;
266 }
267 async ensureAvailable(size: number): Promise<void> {
268 ASSERT(await this.ensureAvailableOrFalse(size), "End of stream");
269 }
270
271 #collectLimitedView(length: number): DataView {
272 ASSERT(this.available >= length);
273 const buffer = this.#buffer;
274 let chunk = UNWRAP(buffer[0]);
275 let offset = this.#offset;
276 if (chunk.byteLength - offset >= length) {
277 this.#shiftOffset(length);
278 return new DataView(chunk.buffer, chunk.byteOffset + offset, length);
279 }
280 let x = 0;
281 while (true) {
282 for (; offset < chunk.byteLength && x < length; offset += 1, x += 1)
283 shared[x] = chunk[offset]!;
284 if (x === length) {
285 if (offset === chunk.byteLength) {
286 ASSERT(buffer.shift());
287 this.#offset = 0;
288 } else this.#offset = offset;
289 this.#available -= length;
290 return sharedView;
291 }
292 ASSERT(buffer.shift());
293 chunk = UNWRAP(buffer[0]);
294 offset = 0;
295 }
296 }
297 #collectBuffer(length: number) {
298 ASSERT(this.available >= length);
299 this.#available -= length;
300 const buffer = this.#buffer;
301 let offset = this.#offset;
302
303 // slice an existing data view
304 if (buffer[0] && buffer[0].byteLength - offset >= length) {
305 const first = buffer[0];
306 if (first.byteLength - offset === length) {
307 this.#offset = 0;
308 ASSERT(buffer.shift());
309 return offset ? first.subarray(offset) : first;
310 }
311 this.#offset += length;
312 return first.subarray(offset, offset + length);
313 }
314
315 // join by allocating a new buffer
316 const joined = new Uint8Array(length);
317 let i = 0;
318 do {
319 const chunk = UNWRAP(buffer[0]);
320 const { byteLength } = chunk;
321 const copyLen = Math.min(byteLength, length);
322 joined.set(copyLen === byteLength ? chunk : chunk.subarray(offset, offset + copyLen), i);
323 i += copyLen;
324 length -= copyLen;
325 if (copyLen === byteLength) ASSERT(buffer.shift());
326 if (length === 0) this.#offset = copyLen;
327 } while (length > 0);
328 return joined;
329 }
330 #shiftOffset(offset: number) {
331 const newOffset = (this.#offset += offset);
332 this.#available -= offset;
333 if (UNWRAP(this.#buffer[0]).byteLength === newOffset) {
334 this.#buffer.shift();
335 this.#offset = 0;
336 } else {
337 this.#offset = newOffset;
338 }
339 }
340
341 async u8(): Promise<number> {
342 const int = await this.peekU8();
343 this.#shiftOffset(1);
344 return int;
345 }
346
347 async peekU8(): Promise<number> {
348 await this.ensureAvailable(1);
349 return UNWRAP(UNWRAP(this.#buffer[0])[this.#offset]);
350 }
351
352 async i8(): Promise<number> {
353 await this.ensureAvailable(2);
354 return this.#collectLimitedView(1).getInt8(0);
355 }
356
357 async u16(): Promise<number> {
358 await this.ensureAvailable(2);
359 return this.#collectLimitedView(2).getUint16(0, true);
360 }
361
362 async i16(): Promise<number> {
363 await this.ensureAvailable(2);
364 return this.#collectLimitedView(2).getInt16(0, true);
365 }
366
367 async u32(): Promise<number> {
368 await this.ensureAvailable(4);
369 return this.#collectLimitedView(4).getUint32(0, true);
370 }
371
372 async i32(): Promise<number> {
373 await this.ensureAvailable(4);
374 return this.#collectLimitedView(4).getInt32(0, true);
375 }
376
377 async u64(): Promise<bigint> {
378 await this.ensureAvailable(8);
379 return this.#collectLimitedView(8).getBigUint64(0, true);
380 }
381
382 async i64(): Promise<bigint> {
383 await this.ensureAvailable(8);
384 return this.#collectLimitedView(8).getBigInt64(0, true);
385 }
386
387 async f16(): Promise<number> {
388 await this.ensureAvailable(2);
389 return this.#collectLimitedView(2).getFloat16(0, true);
390 }
391
392 async f32(): Promise<number> {
393 await this.ensureAvailable(4);
394 return this.#collectLimitedView(4).getFloat32(0, true);
395 }
396
397 async f64(): Promise<number> {
398 await this.ensureAvailable(8);
399 return this.#collectLimitedView(8).getFloat64(0, true);
400 }
401
402 async varUint(): Promise<number> {
403 let result = 0;
404 let shift = 1;
405 let byte;
406 do {
407 byte = await this.u8();
408 result += (byte & 0x7f) * shift;
409 shift *= 128;
410 } while (byte & 0x80);
411 return result;
412 }
413
414 async stringWithLength(): Promise<string> {
415 const len = await this.varUint();
416 return string.decodeUtf8(await this.readExactly(len));
417 }
418
419 /** Read up to a maximum number of bytes */
420 async readUpTo(size: number): Promise<Uint8Array> {
421 await this.ensureAvailableOrFalse(size);
422 return this.#collectBuffer(Math.min(size, this.#available));
423 }
424 async readExactly(size: number): Promise<Uint8Array> {
425 await this.ensureAvailable(size);
426 return this.#collectBuffer(size);
427 }
428
429 [Symbol.dispose]() {
430 this.releaseLock();
431 }
432}
433
434/** easily create a readable stream with a writer */
435export function bufferedReadableStream(): [
436 ReadableStream,
437 BufferedWriter,
438 ReadableStreamDefaultController<Uint8Array>,
439] {
440 let unbuffered: ReadableStreamDefaultController<Uint8Array> | null = null;
441 const reader = new ReadableStream<Uint8Array>({
442 start: (controller) => (unbuffered = controller),
443 });
444 ASSERT(unbuffered);
445 return [reader, new BufferedWriter(unbuffered), unbuffered];
446}
447
448/** easily create a readable stream with a writer */
449export function readWritePair(): [ReadableStream, ReadableStreamDefaultController] {
450 let unbuffered: ReadableStreamDefaultController<Uint8Array> | null = null;
451 const reader = new ReadableStream<Uint8Array>({
452 start: (controller) => (unbuffered = controller),
453 });
454 ASSERT(unbuffered);
455 return [reader, unbuffered];
456}
457
458/**
459 * decode utf-8 and yield each line without its ending, which is any of `\n`,
460 * `\r\n`, or the lone `\r` that command line programs redraw a status with.
461 */
462export async function* lines(readable: ReadableStream<Uint8Array>): AsyncGenerator<string> {
463 const decoder = new TextDecoder();
464 let rest = "";
465 let endsInCarriageReturn = false;
466
467 for await (const bytes of readable) {
468 let chunk = decoder.decode(bytes, { stream: true });
469 if (chunk === "") continue;
470 // the `\n` of a `\r\n` that arrived in two chunks
471 if (endsInCarriageReturn && chunk[0] === "\n") chunk = chunk.slice(1);
472 endsInCarriageReturn = chunk.endsWith("\r");
473
474 const parts = (rest + chunk).split(/\r\n|\r|\n/);
475 rest = UNWRAP(parts.pop());
476 yield* parts;
477 }
478
479 rest += decoder.decode();
480 if (rest) yield rest;
481}
482
483import { ASSERT, UNWRAP } from "./assert.ts";
484import * as string from "./string.ts";