Node.js EventEmitter Patterns

The EventEmitter is central to Node.js architecture. Here's how to use it effectively.

EventEmitter Basics#

const EventEmitter = require('events');

// Create an emitter
const emitter = new EventEmitter();

// Listen for events
emitter.on('message', (data) => {
  console.log('Received:', data);
});

// Emit events
emitter.emit('message', 'Hello, World!');

// One-time listener
emitter.once('connect', () => {
  console.log('Connected (runs once)');
});

// Remove listener
const handler = (data) => console.log(data);
emitter.on('data', handler);
emitter.off('data', handler); // or removeListener

// Remove all listeners
emitter.removeAllListeners('data');
emitter.removeAllListeners(); // All events

Creating Custom Emitters#

const EventEmitter = require('events');

// Class-based emitter
class Database extends EventEmitter {
  constructor(config) {
    super();
    this.config = config;
    this.connected = false;
  }

  async connect() {
    this.emit('connecting');

    try {
      await this.doConnect();
      this.connected = true;
      this.emit('connected');
    } catch (error) {
      this.emit('error', error);
    }
  }

  async query(sql) {
    this.emit('query', sql);

    try {
      const result = await this.execute(sql);
      this.emit('result', result);
      return result;
    } catch (error) {
      this.emit('error', error);
      throw error;
    }
  }

  async doConnect() {
    // Connection logic
  }

  async execute(sql) {
    // Query execution
  }
}

// Usage
const db = new Database({ host: 'localhost' });

db.on('connecting', () => console.log('Connecting...'));
db.on('connected', () => console.log('Connected!'));
db.on('query', (sql) => console.log('Query:', sql));
db.on('error', (err) => console.error('Error:', err));

await db.connect();
await db.query('SELECT * FROM users');

Job Queue Pattern#

const EventEmitter = require('events');

class JobQueue extends EventEmitter {
  constructor(concurrency = 1) {
    super();
    this.concurrency = concurrency;
    this.queue = [];
    this.running = 0;
  }

  add(job) {
    this.queue.push(job);
    this.emit('job:added', job);
    this.process();
  }

  async process() {
    while (this.running < this.concurrency && this.queue.length > 0) {
      const job = this.queue.shift();
      this.running++;
      this.emit('job:started', job);

      try {
        const result = await job.execute();
        this.emit('job:completed', { job, result });
      } catch (error) {
        this.emit('job:failed', { job, error });
      } finally {
        this.running--;
        this.process();
      }
    }

    if (this.running === 0 && this.queue.length === 0) {
      this.emit('queue:empty');
    }
  }
}

// Usage
const queue = new JobQueue(3);

queue.on('job:started', ({ id }) => console.log(`Job ${id} started`));
queue.on('job:completed', ({ job, result }) => console.log(`Job ${job.id} done`));
queue.on('job:failed', ({ job, error }) => console.error(`Job ${job.id} failed`));
queue.on('queue:empty', () => console.log('All jobs complete'));

queue.add({ id: 1, execute: async () => 'Result 1' });
queue.add({ id: 2, execute: async () => 'Result 2' });

WebSocket-like Pattern#

const EventEmitter = require('events');

class Connection extends EventEmitter {
  constructor(id) {
    super();
    this.id = id;
    this.connected = false;
  }

  open() {
    this.connected = true;
    this.emit('open');
  }

  send(message) {
    if (!this.connected) {
      this.emit('error', new Error('Not connected'));
      return;
    }
    this.emit('send', message);
  }

  receive(message) {
    this.emit('message', message);
  }

  close(code, reason) {
    this.connected = false;
    this.emit('close', { code, reason });
  }
}

class ConnectionManager extends EventEmitter {
  constructor() {
    super();
    this.connections = new Map();
  }

  add(connection) {
    this.connections.set(connection.id, connection);

    connection.on('message', (msg) => {
      this.emit('message', { connection, message: msg });
    });

    connection.on('close', () => {
      this.connections.delete(connection.id);
      this.emit('disconnect', connection);
    });

    this.emit('connect', connection);
  }

  broadcast(message, exclude = []) {
    for (const [id, conn] of this.connections) {
      if (!exclude.includes(id)) {
        conn.send(message);
      }
    }
  }
}

Event Bus Pattern#

const EventEmitter = require('events');

class EventBus extends EventEmitter {
  constructor() {
    super();
    this.setMaxListeners(100);
  }

  // Namespaced events
  publish(namespace, event, data) {
    const fullEvent = `${namespace}:${event}`;
    this.emit(fullEvent, data);
    this.emit('*', { namespace, event, data });
  }

  subscribe(namespace, event, handler) {
    const fullEvent = `${namespace}:${event}`;
    this.on(fullEvent, handler);
    return () => this.off(fullEvent, handler);
  }

  // Wildcard subscription
  subscribeAll(handler) {
    this.on('*', handler);
    return () => this.off('*', handler);
  }
}

// Global event bus
const eventBus = new EventBus();

// Module A
const unsubscribe = eventBus.subscribe('users', 'created', (user) => {
  console.log('User created:', user);
});

// Module B
eventBus.publish('users', 'created', { id: 1, name: 'Alice' });

// Cleanup
unsubscribe();

State Machine with Events#

const EventEmitter = require('events');

class StateMachine extends EventEmitter {
  constructor(config) {
    super();
    this.states = config.states;
    this.state = config.initial;
  }

  transition(event) {
    const currentStateConfig = this.states[this.state];
    const nextState = currentStateConfig?.on?.[event];

    if (!nextState) {
      this.emit('transition:invalid', { from: this.state, event });
      return false;
    }

    const prevState = this.state;
    this.state = nextState;

    this.emit('transition', { from: prevState, to: nextState, event });
    this.emit(`enter:${nextState}`, { from: prevState });
    this.emit(`exit:${prevState}`, { to: nextState });

    return true;
  }

  getState() {
    return this.state;
  }

  can(event) {
    return !!this.states[this.state]?.on?.[event];
  }
}

// Usage
const orderMachine = new StateMachine({
  initial: 'pending',
  states: {
    pending: { on: { PAY: 'paid', CANCEL: 'cancelled' } },
    paid: { on: { SHIP: 'shipped', REFUND: 'refunded' } },
    shipped: { on: { DELIVER: 'delivered' } },
    delivered: {},
    cancelled: {},
    refunded: {},
  },
});

orderMachine.on('transition', ({ from, to, event }) => {
  console.log(`Order: ${from} -> ${to} (${event})`);
});

orderMachine.on('enter:shipped', () => {
  console.log('Send shipping notification');
});

orderMachine.transition('PAY');
orderMachine.transition('SHIP');

Error Handling#

const EventEmitter = require('events');

class SafeEmitter extends EventEmitter {
  // Emit with error handling
  safeEmit(event, ...args) {
    try {
      this.emit(event, ...args);
    } catch (error) {
      this.emit('error', error);
    }
  }

  // Async emit with error collection
  async emitAsync(event, ...args) {
    const listeners = this.listeners(event);
    const errors = [];

    for (const listener of listeners) {
      try {
        await listener(...args);
      } catch (error) {
        errors.push(error);
      }
    }

    if (errors.length > 0) {
      this.emit('errors', { event, errors });
    }
  }
}

// Always handle error events
const emitter = new EventEmitter();

emitter.on('error', (error) => {
  console.error('EventEmitter error:', error);
});

// Or use captureRejections
const asyncEmitter = new EventEmitter({ captureRejections: true });

asyncEmitter.on('data', async () => {
  throw new Error('Async error');
});

// This gets emitted as 'error' event
asyncEmitter.emit('data');

Memory Management#

const EventEmitter = require('events');

// Increase max listeners if needed
const emitter = new EventEmitter();
emitter.setMaxListeners(20); // Default is 10

// Check for leaks
console.log(emitter.listenerCount('event'));

// Properly cleanup
class Component extends EventEmitter {
  constructor(eventBus) {
    super();
    this.eventBus = eventBus;
    this.handlers = [];
  }

  initialize() {
    const handler = this.onData.bind(this);
    this.eventBus.on('data', handler);
    this.handlers.push({ event: 'data', handler });
  }

  onData(data) {
    // Handle data
  }

  destroy() {
    // Cleanup all handlers
    for (const { event, handler } of this.handlers) {
      this.eventBus.off(event, handler);
    }
    this.handlers = [];
    this.removeAllListeners();
  }
}

// Auto-cleanup with AbortController
const controller = new AbortController();

emitter.on('event', handler, { signal: controller.signal });

// Later, cleanup
controller.abort();

Typed Events (TypeScript)#

import { EventEmitter } from 'events';

interface UserEvents {
  'user:created': (user: User) => void;
  'user:updated': (user: User, changes: Partial<User>) => void;
  'user:deleted': (userId: string) => void;
}

class TypedEmitter<T extends Record<string, (...args: any[]) => void>> {
  private emitter = new EventEmitter();

  on<K extends keyof T>(event: K, listener: T[K]): this {
    this.emitter.on(event as string, listener);
    return this;
  }

  off<K extends keyof T>(event: K, listener: T[K]): this {
    this.emitter.off(event as string, listener);
    return this;
  }

  emit<K extends keyof T>(event: K, ...args: Parameters<T[K]>): boolean {
    return this.emitter.emit(event as string, ...args);
  }
}

// Usage
const userEvents = new TypedEmitter<UserEvents>();

userEvents.on('user:created', (user) => {
  console.log(user.name); // Typed!
});

userEvents.emit('user:created', { id: '1', name: 'Alice' });

Best Practices#

Design:
✓ Use semantic event names
✓ Document event payloads
✓ Keep events focused
✓ Consider event versioning

Memory:
✓ Remove listeners when done
✓ Watch maxListeners warnings
✓ Use once() for one-time events
✓ Clean up in destroy methods

Error Handling:
✓ Always add error listeners
✓ Handle async errors properly
✓ Use captureRejections
✓ Don't throw in listeners

Performance:
✓ Don't over-emit
✓ Batch related events
✓ Consider throttling
✓ Profile with many listeners

The EventEmitter pattern enables loose coupling and flexible architectures. Use it for async communication, state changes, and plugin systems. Always handle errors, manage memory properly, and document your event contracts.