Node.js EventEmitter - Memory Leak from Anonymous Listeners

// Pull EventEmitter from Node's built-in 'events' module — no install needed
const EventEmitter = require('events');

// Create a direct instance — acceptable for demos and simple scripts
const notificationHub = new EventEmitter();

// Register a listener for the 'userSignedUp' event.
// .on() means: every time this event fires, run this callback.
// The listener stays registered until explicitly removed with .off().
notificationHub.on('userSignedUp', (userData) => {
  // This runs synchronously when the event is emitted —
  // the emit() caller waits for this to complete before continuing.
  console.log(`[Email Service] Sending welcome email to: ${userData.email}`);
});

// Register a second listener for the SAME event.
// Both listeners run on every emit(), in registration order.
notificationHub.on('userSignedUp', (userData) => {
  console.log(`[Analytics] Recording new signup for user ID: ${userData.id}`);
});

// .once() registers a listener that fires exactly one time, then removes itself.
// Node.js handles the removal automatically — you don't need to track it.
// Perfect for one-off setup tasks, first-connection initialization, or
// any situation where reacting more than once would be incorrect.
notificationHub.once('userSignedUp', (userData) => {
  console.log(`[Promo Service] Sending first-signup discount to: ${userData.email}`);
});

console.log('--- First signup ---');
// .emit() fires the event synchronously — all listeners complete
// before the line after this one executes.
notificationHub.emit('userSignedUp', { id: 101, email: 'alex@example.com' });

console.log('\n--- Second signup ---');
// The .once() listener does NOT fire this time — it was auto-removed
// after the first emission. The two .on() listeners still fire.
notificationHub.emit('userSignedUp', { id: 102, email: 'jordan@example.com' });

// listenerCount() is useful for health monitoring — a count that grows
// over time on a long-lived emitter is a memory leak indicator.
console.log(`\nActive listeners: ${notificationHub.listenerCount('userSignedUp')}`);
console.log('// Expected: 2 — the .once() listener was auto-removed after first fire');

const EventEmitter = require('events');

// OrderProcessor extends EventEmitter — it IS an event emitter,
// not something that holds a reference to one.
// This means external code can call processor.on(), processor.once(),
// processor.off() directly, treating the processor as the event source it is.
class OrderProcessor extends EventEmitter {
  constructor() {
    super(); // Required when extending EventEmitter

    // Error listener registered in the constructor so it is always present,
    // regardless of how external code uses this class.
    this.on('error', (err) => {
      console.error(`[OrderProcessor] Internal error: ${err.message}`);
    });

    this.pendingOrders = new Map();
  }

  placeOrder(order) {
    this.pendingOrders.set(order.id, order);
    console.log(`[OrderProcessor] Processing order #${order.id}...`);

    // emit() here — the processor announces what happened.
    // It does not know or care whether Logger, Inventory, or anyone else
    // is listening. Adding a new downstream service means adding a listener
    // externally — zero changes to this class.
    this.emit('orderPlaced', order);
  }

  processPayment(order) {
    // Business rule: single transactions above $500 require manual review
    const approved = order.total < 500;

    if (approved) {
      this.pendingOrders.delete(order.id);
      // Emit enriched data — add server-side timestamp so listeners
      // don't need to generate their own.
      this.emit('paymentConfirmed', {
        ...order,
        confirmedAt: new Date().toISOString()
      });
    } else {
      this.emit('paymentFailed', {
        order,
        reason: 'Amount exceeds single-transaction limit',
        requiresManualReview: true
      });
    }
  }
}

// --- Wire up external services as listeners ---
// None of these services import or modify OrderProcessor.
// They observe it from the outside.

const processor = new OrderProcessor();

// Audit logger — listens to placement for a complete audit trail
processor.on('orderPlaced', (order) => {
  console.log(`[Logger] ORDER PLACED — ID: ${order.id}, Item: ${order.item}, Total: $${order.total}`);
});

// Inventory service — only needs to know about confirmed payments
processor.on('paymentConfirmed', (order) => {
  console.log(`[Inventory] Reserving stock for: ${order.item} (Order #${order.id})`);
});

// Email service — handles both success and failure notifications
processor.on('paymentConfirmed', (order) => {
  console.log(`[Email] Confirmation sent for Order #${order.id} — confirmed at ${order.confirmedAt}`);
});

processor.on('paymentFailed', ({ order, reason, requiresManualReview }) => {
  console.log(`[Email] Failure notice for Order #${order.id}: "${reason}"`);
  if (requiresManualReview) {
    console.log(`[Email] Flagged for manual review queue`);
  }
});

// --- Simulate order flow ---

const order1 = { id: 'ORD-001', item: 'Mechanical Keyboard', total: 120 };
processor.placeOrder(order1);
processor.processPayment(order1);

console.log('');

const order2 = { id: 'ORD-002', item: 'Ultra-Wide Monitor', total: 850 };
processor.placeOrder(order2);
processor.processPayment(order2);

const EventEmitter = require('events');

class DataPipeline extends EventEmitter {
  constructor() {
    super();

    // setMaxListeners is appropriate when your architecture genuinely needs
    // more than 10 listeners — for example, a pipeline that fan-outs to
    // many downstream consumers. Set it explicitly and comment the reason
    // so the next engineer knows this was a deliberate choice, not an oversight.
    this.setMaxListeners(20);

    // Error listener in the constructor — always present, no exceptions.
    // In production, replace console.error with your monitoring service:
    // Sentry.captureException(err), datadogLogs.error(err.message), etc.
    this.on('error', (err) => {
      console.error(`[DataPipeline] Error: ${err.message}`);
    });
  }

  fetchData(sourceUrl) {
    console.log(`[Pipeline] Fetching from: ${sourceUrl}`);

    if (!sourceUrl.startsWith('https://')) {
      // Emitting 'error' is safe here because the constructor registered
      // a handler. Without that handler, this line crashes the process.
      this.emit('error', new Error(`Insecure URL rejected: ${sourceUrl}`));
      return;
    }

    this.emit('dataReceived', { source: sourceUrl, records: 42 });
  }
}

const pipeline = new DataPipeline();

// Named function reference — the ONLY way you can remove this listener later.
// Anonymous: () => {} — registered but unremovable.
// Named:     const handler = () => {} — registered and removable.
const handleDataReceived = (payload) => {
  console.log(`[Consumer] Got ${payload.records} records from ${payload.source}`);
};

pipeline.on('dataReceived', handleDataReceived);

// --- Demonstrate error handling ---
pipeline.fetchData('http://insecure-api.example.com/data'); // triggers error
pipeline.fetchData('https://secure-api.example.com/data');  // succeeds

console.log(`\nListeners before cleanup: ${pipeline.listenerCount('dataReceived')}`);

// .off() requires the exact same function reference used in .on().
// A different function with identical body will not match — Node.js
// compares references, not source code.
pipeline.off('dataReceived', handleDataReceived);

console.log(`Listeners after cleanup: ${pipeline.listenerCount('dataReceived')}`);

// removeAllListeners() with an event name removes all listeners for that event.
// Without an argument it removes ALL listeners for ALL events — including
// the error handler, which means the next error will crash the process.
// Always pass an event name unless you are deliberately tearing down the emitter.
pipeline.removeAllListeners('dataReceived');
console.log(`dataReceived listeners after removeAll: ${pipeline.listenerCount('dataReceived')}`);
console.log(`error listeners still present: ${pipeline.listenerCount('error')}`);

const EventEmitter = require('events');

// ============================================================
// ANTI-PATTERN: New listener registered on every request call.
// This is exactly the pattern from the production incident above.
// ============================================================
function handleRequestAntiPattern(emitter, req, res) {
  // A new anonymous function is created and registered on every call.
  // It captures 'res' in its closure.
  // It can never be removed — anonymous and no .off() call.
  // After 1,000 calls: 1,000 listeners, 1,000 'res' objects in memory.
  emitter.on('responseReady', (data) => {
    res.json(data); // 'res' is retained by the closure
  });
}

// ============================================================
// CORRECT PATTERN: One listener, registered once at startup.
// Per-request state lives in a Map, not in a closure.
// ============================================================
class RequestDispatcher extends EventEmitter {
  constructor() {
    super();
    this.setMaxListeners(5); // Only a few listeners — intentionally bounded
    this.pendingRequests = new Map();
    this.requestCounter = 0;

    // This single listener is registered ONCE in the constructor.
    // It handles dispatch for ALL requests by reading from the Map.
    // Listener count stays at 1 regardless of traffic volume.
    this.on('responseReady', (payload) => {
      const { requestId, data } = payload;
      const res = this.pendingRequests.get(requestId);

      if (!res) {
        // Request timed out or was already handled — discard cleanly
        console.warn(`[Dispatcher] No pending request for ID ${requestId} — discarding`);
        return;
      }

      res.json(data);
      // Remove from Map immediately — allows GC to collect the response object
      this.pendingRequests.delete(requestId);
    });

    // Error listener always present
    this.on('error', (err) => {
      console.error(`[Dispatcher] Error: ${err.message}`);
    });
  }

  handleRequest(req, res) {
    const requestId = ++this.requestCounter;

    // Store the response object in the Map — not in a listener closure
    this.pendingRequests.set(requestId, res);

    // Simulate async work with a timer — in production this would be
    // a database call, an external API request, or a queue message
    setTimeout(() => {
      this.emit('responseReady', {
        requestId,
        data: { status: 'ok', requestId, processed: Date.now() }
      });
    }, 10);

    // Timeout guard: remove from Map after 5 seconds to prevent
    // Map growth from requests that never complete
    setTimeout(() => {
      if (this.pendingRequests.has(requestId)) {
        this.pendingRequests.delete(requestId);
        console.warn(`[Dispatcher] Request ${requestId} timed out — cleaned up`);
      }
    }, 5000);
  }
}

const dispatcher = new RequestDispatcher();

console.log(`Listeners at startup: ${dispatcher.listenerCount('responseReady')}`);

// Simulate 1,000 concurrent requests.
// Listener count must remain constant — if it grows, the pattern is broken.
for (let i = 0; i < 1000; i++) {
  dispatcher.handleRequest({}, { json: (data) => {} });
}

// Allow the async work to complete
setTimeout(() => {
  console.log(`Listeners after 1,000 requests: ${dispatcher.listenerCount('responseReady')}`);
  console.log(`Pending Map entries: ${dispatcher.pendingRequests.size}`);
  console.log('// Expected: 1 listener, 0 pending (all completed within timeout)');
}, 500);

const EventEmitter = require('events');
const { on } = require('events');

// ============================================================
// Pattern 1: Async iteration with events.on()
// Consumes events as a for-await-of loop with natural backpressure.
// The loop body completes before the next event is yielded.
// ============================================================
class MetricsCollector extends EventEmitter {
  constructor() {
    super();
    this.setMaxListeners(20);
    this.on('error', (err) => console.error('[MetricsCollector]', err.message));
  }

  record(name, value) {
    this.emit('metric', { name, value, timestamp: Date.now() });
  }
}

async function processMetricsInBatches(emitter, batchSize = 10, totalLimit = 30) {
  let count = 0;
  let batch = [];

  // events.on() returns an async iterable.
  // Each yield delivers the event arguments as an array.
  // The loop pauses between iterations — producer can emit freely,
  // but consumption is controlled by how fast the loop body runs.
  for await (const [metric] of on(emitter, 'metric')) {
    batch.push(metric);
    count++;

    if (batch.length >= batchSize) {
      // Simulate a batched write — this could be a database insert,
      // a Kafka produce call, or a metrics API flush
      console.log(`[Batch Write] Flushing ${batch.length} metrics to storage (total so far: ${count})`);
      batch = [];
    }

    if (count >= totalLimit) break; // Exit condition — loop terminates cleanly
  }

  if (batch.length > 0) {
    console.log(`[Batch Write] Final flush: ${batch.length} remaining metrics`);
  }

  console.log(`[Done] Consumed ${count} metrics total`);
}

// ============================================================
// Pattern 2: Instrumenting an emitter for observability
// Wraps emit() to add metrics and tracing without modifying
// the emitter class or any of its listeners.
// ============================================================
function instrumentEmitter(emitter, emitterName) {
  const originalEmit = emitter.emit.bind(emitter);

  emitter.emit = function instrumentedEmit(eventName, ...args) {
    // Skip 'newListener' and 'removeListener' to avoid infinite recursion
    if (eventName !== 'newListener' && eventName !== 'removeListener') {
      const listenerCount = emitter.listenerCount(eventName);
      console.log(
        `[Telemetry] ${emitterName} emitting '${eventName}' ` +
        `to ${listenerCount} listener(s) at ${new Date().toISOString()}`
      );
      // In production: increment a Datadog metric, add an OpenTelemetry span, etc.
    }
    return originalEmit(eventName, ...args);
  };

  return emitter;
}

// --- Demonstrate async iteration ---
const collector = new MetricsCollector();

// Start the async consumer before emitting
processMetricsInBatches(collector, 10, 30).catch(console.error);

// Emit 30 metrics — the async loop consumes them in batches
for (let i = 0; i < 30; i++) {
  collector.record('api.latency', Math.round(Math.random() * 200));
}

// --- Demonstrate instrumentation ---
console.log('\n--- Instrumented emitter ---');
const instrumentedCollector = instrumentEmitter(new MetricsCollector(), 'MetricsCollector');
instrumentedCollector.on('metric', (m) => console.log(`[Consumer] Received: ${m.name} = ${m.value}`));
instrumentedCollector.record('http.requests', 1542);