JavaScript Streams API: Process Data Incrementally

The Streams API lets you process data piece by piece as it arrives, rather than waiting for the entire payload to download or load into memory. This matters when you are dealing with large files, real-time data feeds, or any scenario where memory efficiency and time-to-first-byte are important.

The Three Stream Types

The Streams API has three core primitives:

• ReadableStream - a source of data you can read from
• WritableStream - a destination you can write data to
• TransformStream - a stream that transforms data as it passes through

Every fetch() response body is already a ReadableStream. You have been using streams without realizing it:

const response = await fetch("/api/large-dataset");

// response.body is a ReadableStream
const reader = response.body.getReader();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  console.log(`Received ${value.byteLength} bytes`);
}

Reading Streams

The Reader Pattern

The most common way to consume a ReadableStream is with a reader:

async function readStream(stream) {
  const reader = stream.getReader();
  const chunks = [];

  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      chunks.push(value);
    }
  } finally {
    reader.releaseLock();
  }

  return chunks;
}

Async Iteration

ReadableStreams support async iteration, which is more ergonomic:

async function readStreamAsync(stream) {
  const chunks = [];

  for await (const chunk of stream) {
    chunks.push(chunk);
  }

  return chunks;
}

Creating Custom ReadableStreams

You can create your own data sources:

function createCounterStream(max, intervalMs = 100) {
  let count = 0;

  return new ReadableStream({
    start(controller) {
      const timer = setInterval(() => {
        if (count >= max) {
          controller.close();
          clearInterval(timer);
          return;
        }
        controller.enqueue(count);
        count++;
      }, intervalMs);
    },

    cancel() {
      console.log("Stream cancelled");
    },
  });
}

const stream = createCounterStream(10);

for await (const value of stream) {
  console.log(value); // 0, 1, 2, ... 9
}

A more practical example, streaming lines from a large string:

function createLineStream(text) {
  const lines = text.split("\n");
  let index = 0;

  return new ReadableStream({
    pull(controller) {
      if (index >= lines.length) {
        controller.close();
        return;
      }
      controller.enqueue(lines[index]);
      index++;
    },
  });
}

The pull method is called when the consumer asks for more data. The stream only generates data when needed, which is more memory-efficient than start with timers.

TransformStreams

TransformStreams sit between a readable and writable stream, modifying data as it flows through:

function createUpperCaseTransform() {
  return new TransformStream({
    transform(chunk, controller) {
      controller.enqueue(chunk.toUpperCase());
    },
  });
}

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

for await (const chunk of uppercased) {
  console.log(chunk);
}

JSON Line Parser

A practical transform that parses newline-delimited JSON:

function createJsonLineParser() {
  let buffer = "";

  return new TransformStream({
    transform(chunk, controller) {
      buffer += chunk;
      const lines = buffer.split("\n");
      buffer = lines.pop();

      for (const line of lines) {
        if (line.trim()) {
          try {
            controller.enqueue(JSON.parse(line));
          } catch (e) {
            controller.error(new Error(`Invalid JSON: ${line}`));
          }
        }
      }
    },

    flush(controller) {
      if (buffer.trim()) {
        try {
          controller.enqueue(JSON.parse(buffer));
        } catch (e) {
          controller.error(new Error(`Invalid JSON: ${buffer}`));
        }
      }
    },
  });
}

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

for await (const event of events) {
  handleEvent(event);
}

The flush method is called when the input stream closes, letting you process any remaining buffered data.

Practical: Download Progress

One of the most common stream use cases is tracking download progress:

async function fetchWithProgress(url, onProgress) {
  const response = await fetch(url);
  const contentLength = Number(response.headers.get("content-length"));
  let receivedBytes = 0;

  const progressStream = new TransformStream({
    transform(chunk, controller) {
      receivedBytes += chunk.byteLength;
      if (contentLength) {
        onProgress({
          loaded: receivedBytes,
          total: contentLength,
          percent: Math.round((receivedBytes / contentLength) * 100),
        });
      }
      controller.enqueue(chunk);
    },
  });

  const reader = response.body
    .pipeThrough(progressStream)
    .getReader();

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

  const blob = new Blob(chunks);
  return blob;
}

const file = await fetchWithProgress(
  "/downloads/large-file.zip",
  ({ percent }) => {
    progressBar.style.width = `${percent}%`;
    progressText.textContent = `${percent}%`;
  }
);

Practical: Streaming Server-Sent Events

Process SSE streams manually for more control than the EventSource API:

async function streamSSE(url, handlers) {
  const response = await fetch(url);
  const reader = response.body
    .pipeThrough(new TextDecoderStream())
    .getReader();

  let buffer = "";

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

    buffer += value;
    const events = buffer.split("\n\n");
    buffer = events.pop();

    for (const eventText of events) {
      const lines = eventText.split("\n");
      const event = {};

      for (const line of lines) {
        if (line.startsWith("event:")) {
          event.type = line.slice(6).trim();
        } else if (line.startsWith("data:")) {
          event.data = (event.data || "") + line.slice(5).trim();
        } else if (line.startsWith("id:")) {
          event.id = line.slice(3).trim();
        }
      }

      const handler = handlers[event.type] || handlers["*"];
      if (handler) {
        handler(event);
      }
    }
  }
}

await streamSSE("/api/stream", {
  message: (e) => console.log("Message:", JSON.parse(e.data)),
  error: (e) => console.error("Error event:", e.data),
  "*": (e) => console.log("Unknown event:", e),
});

Piping Streams Together

The pipeThrough and pipeTo methods connect streams into pipelines:

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

await response.body
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(createCsvParser())
  .pipeThrough(createFilter(row => row.status === "active"))
  .pipeThrough(createTransform(row => ({
    name: row.name,
    email: row.email,
  })))
  .pipeTo(new WritableStream({
    write(record) {
      displayRecord(record);
    },
  }));

Each .pipeThrough() adds a processing stage. Backpressure propagates automatically through the entire chain. If the writable stream is slow, the readable stream pauses.

WritableStreams

WritableStreams are the destination end of a pipeline:

function createFileWriter() {
  const chunks = [];

  return new WritableStream({
    write(chunk) {
      chunks.push(chunk);
    },

    close() {
      const blob = new Blob(chunks);
      const url = URL.createObjectURL(blob);
      const a = document.createElement("a");
      a.href = url;
      a.download = "output.txt";
      a.click();
      URL.revokeObjectURL(url);
    },

    abort(reason) {
      console.error("Write aborted:", reason);
      chunks.length = 0;
    },
  });
}

await readableStream.pipeTo(createFileWriter());

Backpressure

Backpressure is the mechanism that prevents a fast producer from overwhelming a slow consumer. The Streams API handles this automatically through the highWaterMark setting:

const stream = new ReadableStream({
  start(controller) {
    for (let i = 0; i < 1000; i++) {
      controller.enqueue(generateLargeChunk());
    }
    controller.close();
  },
}, {
  highWaterMark: 5,
});

When the internal queue reaches the high water mark, the stream signals the source to stop producing. When the consumer reads enough chunks to bring the queue below the mark, production resumes.

For custom streams, you can check controller.desiredSize to know whether to keep producing:

const stream = new ReadableStream({
  pull(controller) {
    while (controller.desiredSize > 0) {
      controller.enqueue(getNextChunk());
    }
  },
});

Teeing Streams

You can split a readable stream into two independent copies:

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

const logPromise = stream1
  .pipeThrough(new TextDecoderStream())
  .pipeTo(createLogWriter());

const processPromise = stream2
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(createJsonLineParser())
  .pipeTo(createProcessor());

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

Both streams receive identical data. This is useful for logging, debugging, or processing the same data in different ways.

Wrapping Up

The Streams API provides a memory-efficient way to process data incrementally. ReadableStreams produce data, TransformStreams modify it in transit, and WritableStreams consume it. Backpressure flows automatically through pipelines built with pipeThrough and pipeTo. The most common practical uses are download progress tracking, processing large files without loading them entirely into memory, and handling streaming server responses. Every fetch() response body is already a stream, so you can start using these patterns with data you are already fetching.