Streams & ReadableStream

Process data incrementally with the WHATWG Streams API ReadableStream, WritableStream, and TransformStream for memory-efficient handling of large or infinite data sources.

Overview

Streams are the answer to a fundamental problem: what do you do when data is too large to hold in memory at once, or when it arrives incrementally over time? The WHATWG Streams API provides a composable, backpressure-aware model for reading, transforming, and writing data in chunks. Unlike Node.js streams (which use events), Web Streams are based on Promises and work in both browsers and modern Node.js (v16+). A ReadableStream wraps a source a network response, a file, a sensor, or a generator and lets consumers pull chunks from it. A WritableStream wraps a sink a file, a WebSocket, an HTTP body and lets producers push chunks into it. A TransformStream sits between them: it has a writable side that accepts input chunks and a readable side that emits transformed output chunks. You pipe them together to build processing pipelines: response.body.pipeThrough(decompress).pipeThrough(decode).pipeTo(display). Backpressure is what makes streams practical at scale. If the consumer reads slower than the producer writes, the stream's internal queue fills up and the producer is signalled to slow down. This prevents memory exhaustion. The highWaterMark on a stream's queuingStrategy controls how large the internal buffer can grow before the producer blocks. Understanding backpressure is essential when building custom sources and sinks ignoring it leads to unbounded memory usage.

Key Points

•ReadableStream wraps a source; pull chunks with a reader obtained from stream.getReader()
•WritableStream wraps a sink; push chunks with a writer obtained from stream.getWriter()
•TransformStream has a writable side (input) and readable side (output) use it to transform chunks inline
•pipeThrough(transform) chains a TransformStream; pipeTo(writable) terminates a pipeline
•Backpressure: the stream signals the producer to pause when the internal queue exceeds highWaterMark
•Byte streams (ReadableByteStream) use BYOB (Bring Your Own Buffer) readers for zero-copy reads
•ReadableStream is async-iterable you can use for-await-of to consume it chunk by chunk
•fetch().body returns a ReadableStream you can stream large responses without buffering the whole body

Examples

Reading a fetch response as a stream progress tracking for large downloads

async function downloadWithProgress(url) {
  const response = await fetch(url);
  const contentLength = response.headers.get('Content-Length');
  const total = contentLength ? parseInt(contentLength, 10) : null;

  const reader = response.body.getReader();
  const chunks = [];
  let received = 0;

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    chunks.push(value);
    received += value.byteLength;

    if (total) {
      const pct = Math.round((received / total) * 100);
      console.log(`Progress: ${pct}% (${received}/${total} bytes)`);
    }
  }

  // Concatenate all chunks into a single buffer
  const buffer = new Uint8Array(received);
  let offset = 0;
  for (const chunk of chunks) {
    buffer.set(chunk, offset);
    offset += chunk.byteLength;
  }

  return buffer;
}

This avoids buffering the entire response in memory each chunk is processed as it arrives.

Custom ReadableStream wrapping an async data source

// Wrap an async generator as a ReadableStream
function asyncGeneratorToStream(gen) {
  return new ReadableStream({
    async pull(controller) {
      const { value, done } = await gen.next();
      if (done) {
        controller.close();
      } else {
        controller.enqueue(value);
      }
    }
  });
}

async function* sensorReadings() {
  for (let i = 0; i < 10; i++) {
    await new Promise(r => setTimeout(r, 100)); // simulate sensor delay
    yield { timestamp: Date.now(), temp: 20 + Math.random() * 5 };
  }
}

// Consume as a stream
const stream = asyncGeneratorToStream(sensorReadings());
const reader = stream.getReader();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  console.log(`${value.timestamp}: ${value.temp.toFixed(1)}°C`);
}

The pull() method is called only when the consumer is ready for the next chunk this is how streams naturally implement backpressure.

TransformStream building a processing pipeline

// A TransformStream that parses newline-delimited JSON (NDJSON)
function ndjsonTransform() {
  let buffer = '';
  return new TransformStream({
    transform(chunk, controller) {
      buffer += new TextDecoder().decode(chunk);
      const lines = buffer.split('\n');
      buffer = lines.pop() ?? ''; // keep the incomplete last line

      for (const line of lines) {
        const trimmed = line.trim();
        if (trimmed) {
          try {
            controller.enqueue(JSON.parse(trimmed));
          } catch {
            controller.error(new SyntaxError(`Invalid JSON: ${trimmed}`));
          }
        }
      }
    },
    flush(controller) {
      if (buffer.trim()) {
        controller.enqueue(JSON.parse(buffer.trim()));
      }
    }
  });
}

// Usage: stream an NDJSON API response
async function streamNDJSON(url) {
  const response = await fetch(url);
  const stream = response.body.pipeThrough(ndjsonTransform());

  for await (const record of stream) {
    console.log('Record:', record);
  }
}

The flush() method is called when the writable side is closed use it to process any remaining buffered data.

Common Mistakes

Calling reader.read() in parallel from multiple consumers a ReadableStream can only have one active reader at a time; tee() it to create two independent streams
Forgetting to release the reader lock (reader.releaseLock()) when done the stream remains locked and cannot be read by any other consumer
Ignoring backpressure by enqueuing without checking the controller's desiredSize always await write() or check desiredSize to respect the consumer's pace
Expecting TransformStream to buffer entire input before producing output each transform call should enqueue output for each chunk, not accumulate the whole stream

Note

ReadableStream is async-iterable in modern environments: `for await (const chunk of stream) { ... }`. This is often the cleanest way to consume a stream, and it correctly handles backpressure the loop body completing naturally signals readiness for the next chunk.

Streams & ReadableStream

Overview

Key Points

Examples

Common Mistakes

Note

Exercises

Knowledge Check

What does `stream.tee()` return?