The Streams API in JavaScript

The Streams API lets you process data incrementally, chunk by chunk, rather than waiting for an entire resource to load into memory. This is essential for handling large files, real-time data, and network responses where you want to start processing before the full payload arrives.

Why streams matter

Without streams, fetching a large file looks like this:

// Loads the ENTIRE response into memory at once
const response = await fetch("/api/large-dataset");
const data = await response.json();
process(data);

For a 500 MB file, this blocks until every byte is downloaded, then parses everything at once. With streams, you can process data as it arrives, reducing memory usage and improving time to first result.

ReadableStream

A ReadableStream represents a source of data that you read from. Fetch responses already provide one via response.body.

Reading a stream manually

const response = await fetch("/api/large-file");
const reader = response.body.getReader();
const decoder = new TextDecoder();

let result = "";

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

  if (done) break;

  // value is a Uint8Array chunk
  result += decoder.decode(value, { stream: true });
  console.log(`Received ${value.length} bytes`);
}

console.log("Complete:", result.length, "characters");

Progress tracking

async function fetchWithProgress(url, onProgress) {
  const response = await fetch(url);
  const contentLength = +response.headers.get("Content-Length");
  const reader = response.body.getReader();

  let receivedBytes = 0;
  const chunks = [];

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

    chunks.push(value);
    receivedBytes += value.length;

    if (contentLength) {
      onProgress(receivedBytes / contentLength);
    }
  }

  // Combine chunks into a single Uint8Array
  const allChunks = new Uint8Array(receivedBytes);
  let position = 0;
  for (const chunk of chunks) {
    allChunks.set(chunk, position);
    position += chunk.length;
  }

  return allChunks;
}

const data = await fetchWithProgress("/api/download", (progress) => {
  console.log(`${(progress * 100).toFixed(1)}% complete`);
});

Creating custom ReadableStreams

You can create streams from any data source using the ReadableStream constructor.

function createCounterStream(limit) {
  let count = 0;

  return new ReadableStream({
    pull(controller) {
      if (count < limit) {
        controller.enqueue(count);
        count++;
      } else {
        controller.close();
      }
    },
  });
}

const stream = createCounterStream(5);
const reader = stream.getReader();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  console.log(value); // 0, 1, 2, 3, 4
}

Timer-based stream

function createTimerStream(intervalMs, count) {
  let emitted = 0;

  return new ReadableStream({
    start(controller) {
      const id = setInterval(() => {
        if (emitted >= count) {
          clearInterval(id);
          controller.close();
          return;
        }

        controller.enqueue({
          tick: emitted,
          timestamp: Date.now(),
        });
        emitted++;
      }, intervalMs);
    },
  });
}

const timer = createTimerStream(1000, 3);
for await (const tick of timer) {
  console.log(tick); // { tick: 0, timestamp: ... } every second
}

WritableStream

A WritableStream represents a destination you write to. Use it to process stream data incrementally.

const writableStream = new WritableStream({
  write(chunk) {
    console.log("Writing chunk:", chunk);
  },
  close() {
    console.log("Stream closed");
  },
  abort(reason) {
    console.error("Stream aborted:", reason);
  },
});

const writer = writableStream.getWriter();
await writer.write("Hello");
await writer.write("World");
await writer.close();

Piping to a WritableStream

Instead of reading manually, pipe a readable stream directly to a writable stream.

const response = await fetch("/api/data");

const logStream = new WritableStream({
  write(chunk) {
    console.log(`Chunk size: ${chunk.length} bytes`);
  },
});

await response.body.pipeTo(logStream);

pipeTo handles backpressure automatically — if the writable stream is slow, the readable stream pauses.

TransformStream

A TransformStream sits between a readable and writable stream, transforming data as it passes through.

function createUppercaseTransform() {
  const decoder = new TextDecoder();
  const encoder = new TextEncoder();

  return new TransformStream({
    transform(chunk, controller) {
      const text = decoder.decode(chunk, { stream: true });
      controller.enqueue(encoder.encode(text.toUpperCase()));
    },
  });
}

const response = await fetch("/api/text");
const uppercased = response.body.pipeThrough(createUppercaseTransform());

const reader = uppercased.getReader();
const decoder = new TextDecoder();

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

JSON line parser

Many APIs send newline-delimited JSON. A transform stream can parse each line incrementally.

function createJsonLineParser() {
  let buffer = "";

  return new TransformStream({
    transform(chunk, controller) {
      buffer += new TextDecoder().decode(chunk, { stream: true });
      const lines = buffer.split("\n");

      // Keep the last (possibly incomplete) line in the buffer
      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));
      }
    },
  });
}

const response = await fetch("/api/events");
const parsed = response.body.pipeThrough(createJsonLineParser());

for await (const event of parsed) {
  console.log("Event:", event);
}

Chaining transforms

Multiple transforms can be chained with pipeThrough.

function createFilterTransform(predicateFn) {
  return new TransformStream({
    transform(chunk, controller) {
      if (predicateFn(chunk)) {
        controller.enqueue(chunk);
      }
    },
  });
}

function createMapTransform(mapFn) {
  return new TransformStream({
    transform(chunk, controller) {
      controller.enqueue(mapFn(chunk));
    },
  });
}

const response = await fetch("/api/events");

const pipeline = response.body
  .pipeThrough(createJsonLineParser())
  .pipeThrough(createFilterTransform((event) => event.type === "error"))
  .pipeThrough(createMapTransform((event) => ({
    message: event.message,
    timestamp: new Date(event.timestamp).toISOString(),
  })));

for await (const errorEvent of pipeline) {
  console.log("Error:", errorEvent.message, "at", errorEvent.timestamp);
}

Teeing a stream

tee() splits a readable stream into two independent branches, both of which receive the same data.

const response = await fetch("/api/data");
const [branch1, branch2] = response.body.tee();

// Process both branches independently
const logPromise = branch1.pipeTo(new WritableStream({
  write(chunk) {
    console.log(`Logging: ${chunk.length} bytes`);
  },
}));

const savePromise = branch2.pipeTo(new WritableStream({
  write(chunk) {
    // Save to IndexedDB, file, etc.
  },
}));

await Promise.all([logPromise, savePromise]);

Summary

The Streams API provides a powerful model for incremental data processing:

• ReadableStream represents a data source. Fetch responses, custom generators, and timers can all produce streams.
• WritableStream represents a data destination. Use pipeTo to connect a readable stream to it.
• TransformStream processes data between a source and destination — parse, filter, map, and transform chunks as they flow.
• Streams handle backpressure automatically, preventing fast producers from overwhelming slow consumers.
• pipeThrough chains multiple transforms into clean data pipelines.
• tee() splits a stream so the same data can be consumed by multiple readers.

Use streams whenever you work with large datasets, real-time data, or any scenario where processing data incrementally is more efficient than loading everything at once.