import { BytesList } from "../bytes/bytes_list.ts";
const CR = "\r".charCodeAt(0);const LF = "\n".charCodeAt(0);
export class LineStream extends TransformStream<Uint8Array, Uint8Array> { #bufs = new BytesList(); #prevHadCR = false;
constructor() { super({ transform: (chunk, controller) => { this.#handle(chunk, controller); }, flush: (controller) => { controller.enqueue(this.#mergeBufs(false)); }, }); }
#handle( chunk: Uint8Array, controller: TransformStreamDefaultController<Uint8Array>, ) { const lfIndex = chunk.indexOf(LF);
if (this.#prevHadCR) { this.#prevHadCR = false; if (lfIndex === 0) { controller.enqueue(this.#mergeBufs(true)); this.#handle(chunk.subarray(1), controller); return; } }
if (lfIndex === -1) { if (chunk.at(-1) === CR) { this.#prevHadCR = true; } this.#bufs.add(chunk); } else { let crOrLfIndex = lfIndex; if (chunk[lfIndex - 1] === CR) { crOrLfIndex--; } this.#bufs.add(chunk.subarray(0, crOrLfIndex)); controller.enqueue(this.#mergeBufs(false)); this.#handle(chunk.subarray(lfIndex + 1), controller); } }
#mergeBufs(prevHadCR: boolean): Uint8Array { const mergeBuf = this.#bufs.concat(); this.#bufs = new BytesList();
if (prevHadCR) { return mergeBuf.subarray(0, -1); } else { return mergeBuf; } }}
interface TextLineStreamOptions { allowCR: boolean;}
export class TextLineStream extends TransformStream<string, string> { #buf = ""; #prevHadCR = false; #allowCR: boolean;
constructor(options?: TextLineStreamOptions) { super({ transform: (chunk, controller) => { this.#handle(chunk, controller); }, flush: (controller) => { controller.enqueue(this.#getBuf(this.#prevHadCR)); if (this.#prevHadCR) { controller.enqueue(""); } }, }); this.#allowCR = options?.allowCR ?? false; }
#handle( chunk: string, controller: TransformStreamDefaultController<string>, ) { const lfIndex = chunk.indexOf("\n"); const crIndex = chunk.indexOf("\r");
if (this.#prevHadCR) { this.#prevHadCR = false; controller.enqueue(this.#getBuf(true)); if (lfIndex === 0) { this.#handle(chunk.slice(1), controller); return; } }
if (lfIndex === -1 && crIndex === -1) { this.#buf += chunk; } else if (lfIndex === -1 && crIndex !== -1) { if (crIndex === (chunk.length - 1)) { this.#buf += chunk; this.#prevHadCR = true; } else if (this.#allowCR) { this.#mergeHandle(chunk, crIndex, crIndex, controller); } else { this.#buf += chunk.slice(0, crIndex + 1); this.#handle(chunk.slice(crIndex + 1), controller); } } else if (lfIndex !== -1 && crIndex === -1) { this.#mergeHandle(chunk, lfIndex, lfIndex, controller); } else { if ((lfIndex - 1) === crIndex) { this.#mergeHandle(chunk, crIndex, lfIndex, controller); } else if (crIndex < lfIndex && this.#allowCR) { this.#mergeHandle(chunk, crIndex, crIndex, controller); } else { this.#mergeHandle(chunk, lfIndex, lfIndex, controller); } } }
#mergeHandle( chunk: string, prevChunkEndIndex: number, newChunkStartIndex: number, controller: TransformStreamDefaultController<string>, ) { this.#buf += chunk.slice(0, prevChunkEndIndex); controller.enqueue(this.#getBuf(false)); this.#handle(chunk.slice(newChunkStartIndex + 1), controller); }
#getBuf(prevHadCR: boolean): string { const buf = this.#buf; this.#buf = "";
if (prevHadCR) { return buf.slice(0, -1); } else { return buf; } }}
export class DelimiterStream extends TransformStream<Uint8Array, Uint8Array> { #bufs = new BytesList(); #delimiter: Uint8Array; #inspectIndex = 0; #matchIndex = 0; #delimLen: number; #delimLPS: Uint8Array;
constructor(delimiter: Uint8Array) { super({ transform: (chunk, controller) => { this.#handle(chunk, controller); }, flush: (controller) => { controller.enqueue(this.#bufs.concat()); }, });
this.#delimiter = delimiter; this.#delimLen = delimiter.length; this.#delimLPS = createLPS(delimiter); }
#handle( chunk: Uint8Array, controller: TransformStreamDefaultController<Uint8Array>, ) { this.#bufs.add(chunk); let localIndex = 0; while (this.#inspectIndex < this.#bufs.size()) { if (chunk[localIndex] === this.#delimiter[this.#matchIndex]) { this.#inspectIndex++; localIndex++; this.#matchIndex++; if (this.#matchIndex === this.#delimLen) { const matchEnd = this.#inspectIndex - this.#delimLen; const readyBytes = this.#bufs.slice(0, matchEnd); controller.enqueue(readyBytes); this.#bufs.shift(this.#inspectIndex); this.#inspectIndex = 0; this.#matchIndex = 0; } } else { if (this.#matchIndex === 0) { this.#inspectIndex++; localIndex++; } else { this.#matchIndex = this.#delimLPS[this.#matchIndex - 1]; } } } }}
export class TextDelimiterStream extends TransformStream<string, string> { #buf = ""; #delimiter: string; #inspectIndex = 0; #matchIndex = 0; #delimLPS: Uint8Array;
constructor(delimiter: string) { super({ transform: (chunk, controller) => { this.#handle(chunk, controller); }, flush: (controller) => { controller.enqueue(this.#buf); }, });
this.#delimiter = delimiter; this.#delimLPS = createLPS(new TextEncoder().encode(delimiter)); }
#handle( chunk: string, controller: TransformStreamDefaultController<string>, ) { this.#buf += chunk; let localIndex = 0; while (this.#inspectIndex < this.#buf.length) { if (chunk[localIndex] === this.#delimiter[this.#matchIndex]) { this.#inspectIndex++; localIndex++; this.#matchIndex++; if (this.#matchIndex === this.#delimiter.length) { const matchEnd = this.#inspectIndex - this.#delimiter.length; const readyString = this.#buf.slice(0, matchEnd); controller.enqueue(readyString); this.#buf = this.#buf.slice(this.#inspectIndex); this.#inspectIndex = 0; this.#matchIndex = 0; } } else { if (this.#matchIndex === 0) { this.#inspectIndex++; localIndex++; } else { this.#matchIndex = this.#delimLPS[this.#matchIndex - 1]; } } } }}
function createLPS(pat: Uint8Array): Uint8Array { const lps = new Uint8Array(pat.length); lps[0] = 0; let prefixEnd = 0; let i = 1; while (i < lps.length) { if (pat[i] == pat[prefixEnd]) { prefixEnd++; lps[i] = prefixEnd; i++; } else if (prefixEnd === 0) { lps[i] = 0; i++; } else { prefixEnd = lps[prefixEnd - 1]; } } return lps;}