Event Bus Pattern

Overview

The Event Bus Pattern enables communication between components without direct dependencies. Instead of calling each other directly, components publish events to a central bus and subscribe to events they care about. The bus routes events to all interested subscribers, decoupling publishers from consumers.

This is the foundation of event-driven architecture. A user registration module publishes UserRegistered; email, analytics, and CRM modules subscribe independently. The registration module never knows these consumers exist.

When to Use

Use the Event Bus Pattern when:

• Multiple components need to react to the same event independently
• You want to add new reactions without modifying the publisher
• Cross-cutting concerns (logging, metrics, auditing) must observe operations
• Components must not have compile-time or runtime dependencies on each other
• You need async processing without blocking the main flow

When to Avoid

• Simple one-to-one communication (direct method call is clearer)
• You need guaranteed delivery and ordering (use a message queue instead)
• Debugging requires tracing exact call chains (event buses obscure the flow)
• Events become a hidden control flow that is hard to reason about

Solution

Python

from typing import Callable, List, Dict, Any
from dataclasses import dataclass
import threading

@dataclass
class Event:
    type: str
    payload: Dict[str, Any]


class EventBus:
    def __init__(self):
        self._subscribers: Dict[str, List[Callable]] = {}
        self._lock = threading.Lock()

    def subscribe(self, event_type: str, handler: Callable):
        with self._lock:
            self._subscribers.setdefault(event_type, []).append(handler)

    def publish(self, event: Event):
        handlers = []
        with self._lock:
            handlers = list(self._subscribers.get(event.type, []))
        for handler in handlers:
            handler(event)

    def unsubscribe(self, event_type: str, handler: Callable):
        with self._lock:
            if handler in self._subscribers.get(event_type, []):
                self._subscribers[event_type].remove(handler)


# Usage
bus = EventBus()

def on_user_registered(event: Event):
    print(f"Send welcome email to {event.payload['email']}")

def on_user_registered_analytics(event: Event):
    print(f"Track signup: {event.payload['user_id']}")

bus.subscribe("UserRegistered", on_user_registered)
bus.subscribe("UserRegistered", on_user_registered_analytics)

bus.publish(Event("UserRegistered", {"user_id": 42, "email": "alice@example.com"}))

Java

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;

class Event {
    private final String type;
    private final Map<String, Object> payload;

    public Event(String type, Map<String, Object> payload) {
        this.type = type;
        this.payload = payload;
    }
    public String getType() { return type; }
    public Map<String, Object> getPayload() { return payload; }
}

class EventBus {
    private final Map<String, List<Consumer<Event>>> subscribers = new ConcurrentHashMap<>();

    public void subscribe(String eventType, Consumer<Event> handler) {
        subscribers.computeIfAbsent(eventType, k -> new CopyOnWriteArrayList<>()).add(handler);
    }

    public void publish(Event event) {
        List<Consumer<Event>> handlers = subscribers.getOrDefault(event.getType(), List.of());
        for (Consumer<Event> handler : handlers) {
            handler.accept(event);
        }
    }

    public void unsubscribe(String eventType, Consumer<Event> handler) {
        subscribers.getOrDefault(eventType, List.of()).remove(handler);
    }
}

// Usage
EventBus bus = new EventBus();

bus.subscribe("UserRegistered", event -> {
    System.out.println("Send welcome email to " + event.getPayload().get("email"));
});

bus.subscribe("UserRegistered", event -> {
    System.out.println("Track signup: " + event.getPayload().get("user_id"));
});

bus.publish(new Event("UserRegistered", Map.of("user_id", 42, "email", "alice@example.com")));

JavaScript

class EventBus {
  constructor() {
    this.subscribers = new Map();
  }

  subscribe(eventType, handler) {
    if (!this.subscribers.has(eventType)) {
      this.subscribers.set(eventType, []);
    }
    this.subscribers.get(eventType).push(handler);

    // Return unsubscribe function
    return () => this.unsubscribe(eventType, handler);
  }

  publish(eventType, payload) {
    const handlers = this.subscribers.get(eventType) || [];
    handlers.forEach(handler => {
      try {
        handler(payload);
      } catch (err) {
        console.error(`Handler failed for ${eventType}:`, err);
      }
    });
  }

  unsubscribe(eventType, handler) {
    const handlers = this.subscribers.get(eventType) || [];
    const idx = handlers.indexOf(handler);
    if (idx !== -1) handlers.splice(idx, 1);
  }
}

// Usage
const bus = new EventBus();

const unsubEmail = bus.subscribe('UserRegistered', (payload) => {
  console.log(`Send welcome email to ${payload.email}`);
});

bus.subscribe('UserRegistered', (payload) => {
  console.log(`Track signup: ${payload.user_id}`);
});

bus.publish('UserRegistered', { user_id: 42, email: 'alice@example.com' });

// Later: unsubEmail(); // Remove specific handler

Explanation

The Event Bus Pattern consists of:

• Event: A lightweight message carrying a type and payload
• Publisher: Code that calls publish() without knowing subscribers
• Subscriber: Code that registers a callback via subscribe()
• Bus: Routes events from publishers to all matching subscribers

Variants

Variant	Delivery	Use Case
Synchronous	Immediate, blocking	In-process UI events
Asynchronous	Queued, non-blocking	High-throughput backends
Prioritized	Ordered by priority	UI frameworks (DOM events bubble)
Filtered	Subscribers define predicates	Large systems with many event types

What Works

• Keep event payloads immutable. Subscribers should not modify shared payload objects.
• Use typed event names. Prefer "OrderPlaced" over "order_event". Use constants or enums.
• Isolate subscriber failures. One failing handler should not prevent others from running. Catch and log exceptions per handler.
• Unsubscribe on cleanup. Memory leaks occur when destroyed components still hold subscriptions.
• Document the event schema. Payload structure is an implicit contract. Document required and optional fields.

Common Mistakes

• Chaining events where A triggers B, which triggers C, which triggers A again. Use event sourcing or sagas for complex workflows.
• Over-using the bus for simple parent-child communication makes code harder to follow than a direct callback.
• Forgetting to unsubscribe causes memory leaks and stale updates from destroyed UI components.
• Synchronous handlers doing I/O blocks the publisher. Offload slow work to background threads or queues.
• Untyped payloads force subscribers to cast and guess field names. Use schema validation or strong typing.

Real-World Examples

Android LocalBroadcastManager

Android’s event bus allows fragments and services to communicate without direct references. Replaced by LiveData but the pattern remains.

Vue.js Event Bus

Vue’s $emit / $on provides component-level event buses. Global state management (Pinia) is preferred for cross-app communication.

Guava EventBus

Google’s Java library provides annotation-driven subscription (@Subscribe) with synchronous and async delivery options.

Advanced Topics

Scenario: Event Bus for Event-Driven Architecture

// Event Bus: pub/sub with typing and middleware
type EventHandler<T = unknown> = (event: T) => void | Promise<void>;

class TypedEventBus {
  private handlers = new Map<string, EventHandler[]>()
  private middleware: ((event: string, data: unknown, next: () => void) => void)[] = [];

  on<T>(event: string, handler: EventHandler<T>): () => void {
    if (!this.handlers.has(event)) this.handlers.set(event, []);
    this.handlers.get(event)!.push(handler as EventHandler);
    return () => {
      const list = this.handlers.get(event);
      if (list) this.handlers.set(event, list.filter(h => h !== handler));
    };
  }

  use(mw: (event: string, data: unknown, next: () => void) => void): void {
    this.middleware.push(mw);
  }

  async emit<T>(event: string, data: T): Promise<void> {
    let idx = 0;
    const runMiddleware = () => {
      if (idx < this.middleware.length) {
        const mw = this.middleware[idx++];
        mw(event, data, runMiddleware);
      } else {
        this.dispatch(event, data);
      }
    };
    runMiddleware();
  }

  private async dispatch(event: string, data: unknown): Promise<void> {
    const list = this.handlers.get(event) || [];
    for (const handler of list) {
      try { await handler(data); }
      catch (err) { console.error(`[BUS] Handler error for ${event}:`, err); }
    }
  }
}

// Usage: middleware for logging and validation
const bus = new TypedEventBus();
bus.use((event, data, next) => { console.log(`[LOG] ${event}`); next(); });
bus.use((event, data, next) => {
  if (!data) { console.error("[VALIDATE] Empty data"); return; }
  next();
});

bus.on("user.created", async (user: User) => {
  await emailService.welcome(user.email);
});
bus.on("user.created", async (user: User) => {
  await analytics.track("signup", { userId: user.id });
});

await bus.emit("user.created", { id: "123", email: "alice@test.com" });
// [LOG] user.created
// -> welcome email sent
// -> analytics tracked

Lessons:

• Event Bus: pub/sub with middleware chain (like Express)
• Middleware: logging, validation, auth before dispatch
• Async handlers: the bus awaits each handler sequentially
• on() returns unsubscribe: automatic cleanup
• In distributed systems: use Kafka/RabbitMQ instead of in-memory bus
• The in-memory bus is for a single process: no persistence or distribution

### Event Bus vs Message Queue: which do I use?

Use in-memory Event Bus to decouple modules within a process: notifications, logging, analytics. Use Message Queue (Kafka, RabbitMQ, SQS) to decouple microservices: persistence, retry, DLQ, ordering. Event Bus is volatile: if the process dies, events are lost. MQ is persistent: survives restarts. For a modular monolith, Event Bus. For microservices, MQ.