Stream API - Transform Streams

A TransformStream connects a writable side to a readable side. Each chunk that enters can be modified, expanded, filtered, or dropped before leaving.

Text decoding as a transform

const response = await fetch('/api/notes');
const textStream = response.body.pipeThrough(new TextDecoderStream());

for await (const text of textStream) {
  editor.append(text);
}

TextDecoderStream handles multi-byte character boundaries, so you never see broken characters at chunk edges.

Building a custom transform

const uppercase = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
  flush(controller) {
    controller.enqueue('\n[end]\n');
  }
});

await response.body
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(uppercase)
  .pipeTo(new WritableStream({
    write(text) { output.append(text); }
  }));

Filtering a stream

const errorsOnly = new TransformStream({
  transform(line, controller) {
    if (line.includes('ERROR')) {
      controller.enqueue(line);
    }
  }
});

source
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(lineSplitter)
  .pipeThrough(errorsOnly)
  .pipeTo(displaySink);

A transform can drop chunks by simply not calling enqueue().

Parsing newline-delimited JSON

let buffer = '';

const ndjson = new TransformStream({
  transform(text, controller) {
    buffer += text;
    const lines = buffer.split('\n');
    buffer = lines.pop();

    for (const line of lines) {
      if (line.trim()) {
        controller.enqueue(JSON.parse(line));
      }
    }
  },
  flush(controller) {
    if (buffer.trim()) controller.enqueue(JSON.parse(buffer));
  }
});

Compression

const compressed = new CompressionStream('gzip');
const response = await fetch('/api/backup');

await response.body
  .pipeThrough(compressed)
  .pipeTo(downloadSink);

Compression and decompression streams are available natively in modern browsers, as is DecompressionStream.

Guidelines

• Keep transforms pure so they are easy to test.
• Flush any buffered remainder in flush().
• Handle parse errors inside the transform so one bad chunk does not silently corrupt output.
• Watch memory if a transform can accumulate chunks.

Back to course overview · Previous: Stream API - Writable Streams · Next: Stream API - Request Object