Event Emitters

Challenge

Your application has multiple components that need to communicate with each other without tight coupling. How do you implement a publish-subscribe pattern?

Explanation

EventEmitter is Node.js’s implementation of the observer pattern. It allows objects to emit named events and other objects to listen for those events, enabling decoupled communication.

EventEmitter Basics

const EventEmitter = require('events');

class MyEmitter extends EventEmitter {}

const emitter = new MyEmitter();

// Listen for an event
emitter.on('start', (data) => {
  console.log('Started with:', data);
});

// Emit an event
emitter.emit('start', { time: Date.now() });

// Remove listener
function handleError(err) {
  console.error('Error:', err);
}

emitter.on('error', handleError);
emitter.off('error', handleError);

// Listen once
emitter.once('connection', () => {
  console.log('First connection only');
});

Solution: Custom Event-Driven Server

const EventEmitter = require('events');
const http = require('http');

class CustomServer extends EventEmitter {
  constructor(port) {
    super();
    this.port = port;
    this.requestCount = 0;
  }

  start() {
    const server = http.createServer((req, res) => {
      this.requestCount++;

      // Emit request event
      this.emit('request', {
        method: req.method,
        url: req.url,
        count: this.requestCount
      });

      res.writeHead(200);
      res.end('OK');
    });

    server.listen(this.port, () => {
      this.emit('started', { port: this.port });
    });
  }
}

// Usage
const server = new CustomServer(3000);

server.on('started', (data) => {
  console.log(`Server started on port ${data.port}`);
});

server.on('request', (data) => {
  console.log(`Request #${data.count}: ${data.method} ${data.url}`);

  if (data.count === 100) {
    server.emit('milestone', { count: 100 });
  }
});

server.on('milestone', (data) => {
  console.log(`Milestone reached: ${data.count} requests`);
});

server.start();

Advanced Event Emitter Patterns

const EventEmitter = require('events');

// Async event handling
class AsyncEmitter extends EventEmitter {
  async emitAsync(event, ...args) {
    const listeners = this.listeners(event);

    for (const listener of listeners) {
      await listener(...args);
    }
  }
}

// Max listeners (prevent memory leaks)
const emitter = new EventEmitter();
emitter.setMaxListeners(10);

emitter.on('warning', () => {
  if (emitter.listenerCount('event') > 5) {
    console.warn('Too many listeners');
  }
});

// Prefix all events
class PrefixedEmitter extends EventEmitter {
  emit(event, ...args) {
    return super.emit(`prefix:${event}`, ...args);
  }

  on(event, listener) {
    return super.on(`prefix:${event}`, listener);
  }
}

// Chaining emitters
class PipedEmitter extends EventEmitter {
  pipe(target) {
    this.on('data', (data) => {
      target.emit('data', data);
    });
    return target;
  }
}

Practical Example: State Machine

const EventEmitter = require('events');

class StateMachine extends EventEmitter {
  constructor() {
    super();
    this.state = 'idle';
  }

  setState(newState) {
    const oldState = this.state;
    this.state = newState;

    this.emit('statechange', { oldState, newState });
    this.emit(`enter:${newState}`);
    this.emit(`leave:${oldState}`);
  }
}

// Usage
const fsm = new StateMachine();

fsm.on('enter:running', () => {
  console.log('Entered running state');
});

fsm.on('leave:idle', () => {
  console.log('Left idle state');
});

fsm.on('statechange', ({ oldState, newState }) => {
  console.log(`State changed: ${oldState} -> ${newState}`);
});

fsm.setState('running');
fsm.setState('idle');

Error Handling

const EventEmitter = require('events');

const emitter = new EventEmitter();

// Error event is special - thrown if no listener
emitter.on('error', (err) => {
  console.error('Handled error:', err.message);
});

// emit with error
emitter.emit('error', new Error('Something went wrong'));

// All emitters should handle errors
class SafeEmitter extends EventEmitter {
  safeEmit(event, ...args) {
    try {
      this.emit(event, ...args);
    } catch (err) {
      this.emit('error', err);
    }
  }
}

Memory Leak Prevention

const EventEmitter = require('events');

// ❌ Potential memory leak
const emitter = new EventEmitter();

function onData(data) {
  console.log(data);
}

// Adding listeners repeatedly
for (let i = 0; i < 1000; i++) {
  emitter.on('data', onData); // Multiple copies!
}

// ✅ Better: Clean up listeners
emitter.removeAllListeners('data');
emitter.on('data', onData); // Only one listener

// ✅ Or use once
emitter.once('data', onData);

// ✅ Check listener count
console.log(emitter.listenerCount('data'));

// ✅ Set max listeners limit
emitter.setMaxListeners(10);