Typed Web Streams Transform Pipeline

export function createLineDecoder(): TransformStream<Uint8Array, string> {
  const decoder = new TextDecoder();
  let buffer = "";
  return new TransformStream<Uint8Array, string>({
    transform(chunk, controller) {
      buffer += decoder.decode(chunk, { stream: true });
      let newline = buffer.indexOf("\n");
      while (newline !== -1) {
        controller.enqueue(buffer.slice(0, newline).replace(/\r$/, ""));
        buffer = buffer.slice(newline + 1);
        newline = buffer.indexOf("\n");
      }
      if (buffer.length > 1_000_000) {
        controller.error(new RangeError("line exceeds the 1 MB limit"));
        buffer = "";
      }
    },
    flush(controller) {
      buffer += decoder.decode();
      if (buffer !== "") controller.enqueue(buffer.replace(/\r$/, ""));
      buffer = "";
    },
  });
}

export async function collectStream<Value>(stream: ReadableStream<Value>): Promise<Value[]> {
  if (!(stream instanceof ReadableStream)) throw new TypeError("stream must be a ReadableStream");
  const values: Value[] = [];
  const reader = stream.getReader();
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) return values;
      values.push(value);
    }
  } finally {
    reader.releaseLock();
  }
}

async function* toChunks(text: string): AsyncGenerator<Uint8Array> {
  const bytes = new TextEncoder().encode(text);
  for (let offset = 0; offset < bytes.length; offset += 7) {
    yield bytes.slice(offset, offset + 7);
  }
}

const source = new ReadableStream<Uint8Array>({
  async start(controller) {
    for await (const chunk of toChunks("alpha\r\nbeta\ngamma")) controller.enqueue(chunk);
    controller.close();
  },
});
const lines = await collectStream(source.pipeThrough(createLineDecoder()));
console.log(lines);

const empty = new ReadableStream<Uint8Array>({
  start(controller) {
    controller.close();
  },
});
console.log(await collectStream(empty.pipeThrough(createLineDecoder())));