Node.js Streams and Events: EventEmitter, pipeline, readline

By Engr. Henry Akpan

The Two Ideas Behind Node.js

Much of Node.js is built on two simple ideas. The first is events: objects announce that something happened ("a request arrived", "the file finished loading"), and any number of listeners react. The second is streams: data is processed piece by piece as it flows, instead of loading everything into memory first. HTTP servers, file reading, network sockets and even process.stdin are all event emitters and streams. This tutorial on Node.js streams and events covers the EventEmitter class, writing your own emitters, readable, writable and transform streams, piping and pipeline, backpressure, and a practical example that processes a large CSV file line by line. Each code block runs on its own with Node.js, and the outputs are real.

The EventEmitter

The events module provides EventEmitter. Register listeners with on, and trigger them with emit, passing any arguments:

const EventEmitter = require("events");

const school = new EventEmitter();

school.on("result", (name, score) => {
  console.log(`${name} scored ${score}`);
});
school.on("result", (name, score) => {
  if (score >= 70) console.log(`  ${name} gets an A`);
});
school.once("closed", () => console.log("Portal closed"));

school.emit("result", "Ada", 85);
school.emit("result", "Musa", 62);
school.emit("closed");
school.emit("closed");               // once: the listener already removed itself
console.log("listeners:", school.listenerCount("result"));

Output:

Ada scored 85
  Ada gets an A
Musa scored 62
Portal closed
listeners: 2

Important points:

  • Listeners run synchronously, in the order they were added, when emit is called.
  • once registers a listener that removes itself after the first call.
  • off (or removeListener) removes a listener; you must pass the same function, so keep a reference to it.
  • Emitting an "error" event with no listener throws the error and can crash the program, so always listen for "error" on emitters that can fail.

Building Your Own Emitter Class

Extend EventEmitter to give your own objects events. This download tracker reports progress:

const EventEmitter = require("events");

class Download extends EventEmitter {
  constructor(sizeKb) {
    super();
    this.sizeKb = sizeKb;
  }
  start(chunkKb) {
    let done = 0;
    while (done < this.sizeKb) {
      done = Math.min(done + chunkKb, this.sizeKb);
      this.emit("progress", Math.round((done / this.sizeKb) * 100));
    }
    this.emit("done", this.sizeKb);
  }
}

const d = new Download(1000);
d.on("progress", pct => console.log(`${pct}%`));
d.on("done", kb => console.log(`Downloaded ${kb} KB`));
d.start(400);

Output:

40%
80%
100%
Downloaded 1000 KB

The tracker doesn't need to know who is listening: a progress bar, a log file and a notification can all subscribe independently. This loose coupling is the main benefit of the event pattern.

Why Node.js Streams?

Reading a 2 GB log file with fs.readFileSync loads all 2 GB into memory at once, which may crash a small server. A stream reads it in small chunks (64 KB by default for files), so memory use stays small and processing can start immediately. There are four kinds of stream:

  • Readable: a source of data, such as fs.createReadStream or an incoming HTTP request.
  • Writable: a destination, such as fs.createWriteStream or an HTTP response.
  • Duplex: both readable and writable, such as a network socket.
  • Transform: a duplex stream that changes data as it passes through, such as compression with zlib.

Every stream is also an EventEmitter, emitting events such as "data", "end", "finish" and "error".

Readable Streams

Readable.from turns any iterable into a readable stream. Readable streams are async iterables, so for await...of is the simplest way to consume them:

const { Readable } = require("stream");

const source = Readable.from(["chunk 1", "chunk 2", "chunk 3"]);

(async () => {
  for await (const chunk of source) {
    console.log("received:", chunk);
  }
  console.log("stream ended");
})();

Output:

received: chunk 1
received: chunk 2
received: chunk 3
stream ended

The older style listens for events instead: stream.on("data", chunk => ...) and stream.on("end", () => ...).

Transform Streams and pipeline

A Transform stream receives chunks, changes them and pushes the results on. stream.pipeline connects streams together, passes data along, handles backpressure and reports errors from any stage in one place:

const { Readable, Transform, Writable, pipeline } = require("stream");

const upper = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  }
});

const collected = [];
const sink = new Writable({
  write(chunk, encoding, callback) {
    collected.push(chunk.toString());
    callback();
  }
});

pipeline(Readable.from(["lagos ", "abuja ", "kano"]), upper, sink, (err) => {
  if (err) console.error("Pipeline failed:", err);
  else console.log("Result:", collected.join(""));
});

Output:

Result: LAGOS ABUJA KANO

The callback passed to transform or write tells the stream that you have finished with the current chunk and are ready for the next.

Backpressure

What if the source produces data faster than the destination can handle it, such as reading from a fast disk and writing to a slow network? Without control, unsent data would pile up in memory. This is the backpressure problem. write() returns false when the destination's internal buffer is full, and the writer should pause until the "drain" event. pipe() and pipeline() do this automatically, which is a big reason to use them instead of copying data by hand with "data" events.

Practical Example: Processing a Large CSV File Line by Line

This script writes a sample results file, then streams it back line by line with the built-in readline module to compute an average without ever holding the whole file in memory. The same code would work on a file of millions of lines.

const fs = require("fs");
const readline = require("readline");

const rows = ["name,score", "Ada,85", "Musa,62", "Chidi,74", "Fatima,91"];
fs.writeFileSync("results.csv", rows.join("\n") + "\n");

(async () => {
  const lines = readline.createInterface({
    input: fs.createReadStream("results.csv", { encoding: "utf8" }),
    crlfDelay: Infinity
  });
  let count = 0, total = 0, best = null;
  for await (const line of lines) {
    if (!line || line.startsWith("name")) continue;   // skip blank lines and the header
    const [name, score] = line.split(",");
    const s = Number(score);
    count++;
    total += s;
    if (!best || s > best.s) best = { name, s };
  }
  console.log(`Students: ${count}, average: ${(total / count).toFixed(1)}`);
  console.log(`Top: ${best.name} (${best.s})`);
  fs.unlinkSync("results.csv");
})();

Output:

Students: 4, average: 78.0
Top: Fatima (91)

Similarly, copying and compressing a large file is a one-line pipeline: pipeline(fs.createReadStream("big.log"), zlib.createGzip(), fs.createWriteStream("big.log.gz"), callback). And in an HTTP server, fs.createReadStream(file).pipe(res) sends a large file to a browser without loading it into memory.

Common Mistakes

  • Forgetting an "error" listener (or a pipeline callback), so one failed stream crashes the process.
  • Using readFileSync on huge files when a stream would do.
  • Copying data with "data" events and write() while ignoring its return value, causing memory to balloon (backpressure).
  • Adding a new listener every time a function runs, which leaks memory; Node warns when an emitter has more than 10 listeners for one event.

Summary

  • EventEmitter lets objects announce events with emit; listeners added with on or once run synchronously in order.
  • Node.js streams process data in chunks: readable, writable, duplex and transform.
  • Readable streams are async iterables, so for await...of consumes them cleanly.
  • pipeline() connects streams, handles backpressure and collects errors.
  • Streams and readline let you process files far larger than memory.

Related Tutorials