Skip to main content

Handling Streams

1. Understanding Node.js Streams

TypeUse
ReadableSource (file, HTTP req)
WritableSink (file, HTTP res)
DuplexBoth (TCP socket)
TransformModify chunks (gzip, hash)

2. Piping Streams to Response

Example: Pipe a file

import { createReadStream } from "node:fs";
app.get("/files/report.csv", (req, res) => {
  res.type("text/csv").attachment("report.csv");
  createReadStream("./reports/report.csv").pipe(res);
});

3. Reading Large Files

Example: Highwatermark

createReadStream("./big.bin", { highWaterMark: 1024 * 1024 }).pipe(res);  // 1 MB chunks

4. Sending Streaming Responses

Example: Stream JSON array

app.get("/users.ndjson", async (req, res) => {
  res.type("application/x-ndjson");
  for await (const user of db.users.cursor()) {
    if (!res.write(JSON.stringify(user) + "\n")) {
      await new Promise(r => res.once("drain", r));
    }
  }
  res.end();
});

5. Handling Stream Errors

Example: pipeline()

import { pipeline } from "node:stream/promises";
app.get("/download", async (req, res, next) => {
  try {
    await pipeline(createReadStream("./big.bin"), res);
  } catch (err) { next(err); }
});

6. Using Transform Streams

Example: Inline gzip

import { createGzip } from "node:zlib";
app.get("/big.json.gz", (req, res) => {
  res.set({ "Content-Type": "application/json", "Content-Encoding": "gzip" });
  pipeline(createReadStream("./big.json"), createGzip(), res).catch(err => res.destroy(err));
});

7. Implementing Backpressure

Example: Honor write() return

function emit(res, chunk) {
  if (!res.write(chunk)) {
    return new Promise(r => res.once("drain", r));
  }
}
Note: pipeline() handles backpressure automatically — prefer it over manual pipe() chains.

8. Streaming File Uploads

Example: Stream upload to S3

import busboy from "busboy";
import { Upload } from "@aws-sdk/lib-storage";

app.post("/upload", (req, res, next) => {
  const bb = busboy({ headers: req.headers, limits: { fileSize: 100 * 1024 * 1024 } });
  bb.on("file", (name, file, info) => {
    const upload = new Upload({
      client: s3,
      params: { Bucket: "uploads", Key: info.filename, Body: file, ContentType: info.mimeType }
    });
    upload.done().then(() => res.json({ ok: true })).catch(next);
  });
  req.pipe(bb);
});

9. Using stream.pipeline() Method

Example: Multi-stage pipeline

import { pipeline } from "node:stream/promises";
import { createReadStream, createWriteStream } from "node:fs";
import { createBrotliCompress } from "node:zlib";

await pipeline(createReadStream("in.txt"), createBrotliCompress(), createWriteStream("out.txt.br"));

10. Implementing Server-Sent Events (SSE)

Example: SSE basics

app.get("/events", (req, res) => {
  res.set({
    "Content-Type":  "text/event-stream",
    "Cache-Control": "no-cache, no-transform",
    "Connection":    "keep-alive"
  });
  res.flushHeaders();

  const tick = setInterval(() => res.write(`data: ${Date.now()}\n\n`), 1000);
  req.on("close", () => clearInterval(tick));
});

Was this article helpful?