import { concat } from "../bytes/concat.ts";import { createLPS } from "./_common.ts";
export type DelimiterDisposition = | "suffix" | "prefix" | "discard" ;
export interface DelimiterStreamOptions { disposition?: DelimiterDisposition;}
export class DelimiterStream extends TransformStream<Uint8Array, Uint8Array> { #bufs: Uint8Array[] = []; #delimiter: Uint8Array; #matchIndex = 0; #delimLPS: Uint8Array | null; #disp: DelimiterDisposition;
constructor( delimiter: Uint8Array, options?: DelimiterStreamOptions, ) { super({ transform: (chunk, controller) => delimiter.length === 1 ? this.#handleChar(chunk, controller) : this.#handle(chunk, controller), flush: (controller) => this.#flush(controller), });
this.#delimiter = delimiter; this.#delimLPS = delimiter.length > 1 ? createLPS(delimiter) : null; this.#disp = options?.disposition ?? "discard"; }
#handle( chunk: Uint8Array, controller: TransformStreamDefaultController<Uint8Array>, ) { const bufs = this.#bufs; const length = chunk.byteLength; const disposition = this.#disp; const delimiter = this.#delimiter; const delimLen = delimiter.length; const lps = this.#delimLPS as Uint8Array; let chunkStart = 0; let matchIndex = this.#matchIndex; let inspectIndex = 0; while (inspectIndex < length) { if (chunk[inspectIndex] === delimiter[matchIndex]) { inspectIndex++; matchIndex++; if (matchIndex === delimLen) { matchIndex = 0; const delimiterStartIndex = inspectIndex - delimLen; const delimitedChunkEnd = disposition === "suffix" ? inspectIndex : delimiterStartIndex; if (delimitedChunkEnd <= 0 && bufs.length === 0) { controller.enqueue(new Uint8Array()); chunkStart = disposition === "prefix" ? 0 : inspectIndex; } else if (delimitedChunkEnd > 0 && bufs.length === 0) { controller.enqueue(chunk.subarray(chunkStart, delimitedChunkEnd)); chunkStart = disposition === "prefix" ? delimiterStartIndex : inspectIndex; } else if (delimitedChunkEnd === 0 && bufs.length > 0) { if (bufs.length === 1) { controller.enqueue(bufs[0]!); } else { controller.enqueue(concat(bufs)); } bufs.length = 0; if (disposition !== "prefix") { chunkStart = inspectIndex; } else { chunkStart = 0; } } else if (delimitedChunkEnd < 0 && bufs.length > 0) { const lastIndex = bufs.length - 1; const last = bufs[lastIndex]!; const lastSliceIndex = last.byteLength + delimitedChunkEnd; const lastSliced = last.subarray(0, lastSliceIndex); if (lastIndex === 0) { controller.enqueue(lastSliced); } else { bufs[lastIndex] = lastSliced; controller.enqueue(concat(bufs)); } bufs.length = 0; if (disposition === "prefix") { bufs.push(last.subarray(lastSliceIndex)); chunkStart = 0; } else { chunkStart = inspectIndex; } } else if (delimitedChunkEnd > 0 && bufs.length > 0) { const chunkSliced = chunk.subarray(chunkStart, delimitedChunkEnd); const result = concat([...bufs, chunkSliced]); bufs.length = 0; controller.enqueue(result); chunkStart = disposition === "prefix" ? delimitedChunkEnd : inspectIndex; } else { throw new Error("unreachable"); } } } else if (matchIndex === 0) { inspectIndex++; } else { matchIndex = lps[matchIndex - 1]!; } } this.#matchIndex = matchIndex; if (chunkStart === 0) { bufs.push(chunk); } else if (chunkStart < length) { bufs.push(chunk.subarray(chunkStart)); } }
#handleChar( chunk: Uint8Array, controller: TransformStreamDefaultController<Uint8Array>, ) { const bufs = this.#bufs; const length = chunk.byteLength; const disposition = this.#disp; const delimiter = this.#delimiter[0]; let chunkStart = 0; let inspectIndex = 0; while (inspectIndex < length) { if (chunk[inspectIndex] === delimiter) { inspectIndex++; const delimitedChunkEnd = disposition === "suffix" ? inspectIndex : inspectIndex - 1; if (delimitedChunkEnd === 0 && bufs.length === 0) { controller.enqueue(new Uint8Array()); chunkStart = disposition === "prefix" ? 0 : 1; } else if (delimitedChunkEnd > 0 && bufs.length === 0) { controller.enqueue(chunk.subarray(chunkStart, delimitedChunkEnd)); chunkStart = disposition === "prefix" ? inspectIndex - 1 : inspectIndex; } else if (delimitedChunkEnd === 0 && bufs.length > 0) { if (bufs.length === 1) { controller.enqueue(bufs[0]!); } else { controller.enqueue(concat(bufs)); } bufs.length = 0; if (disposition !== "prefix") { chunkStart = inspectIndex; } } else if (delimitedChunkEnd > 0 && bufs.length > 0) { const chunkSliced = chunk.subarray(chunkStart, delimitedChunkEnd); const result = concat([...bufs, chunkSliced]); bufs.length = 0; chunkStart = disposition === "prefix" ? delimitedChunkEnd : inspectIndex; controller.enqueue(result); } else { throw new Error("unreachable"); } } else { inspectIndex++; } } if (chunkStart === 0) { bufs.push(chunk); } else if (chunkStart < length) { bufs.push(chunk.subarray(chunkStart)); } }
#flush(controller: TransformStreamDefaultController<Uint8Array>) { const bufs = this.#bufs; const length = bufs.length; if (length === 0) { controller.enqueue(new Uint8Array()); } else if (length === 1) { controller.enqueue(bufs[0]!); } else { controller.enqueue(concat(bufs)); } }}