Observer Pattern: Event-Driven Design & Pub-Sub Implementation

Problem

Modern applications require loose coupling between components. When one object changes state, multiple other objects need to react without tight dependencies. Traditional approaches create brittle code where:

• Components directly reference each other
• Changes cascade unpredictably
• Testing becomes difficult
• Scaling adds complexity
• Adding new listeners requires modifying existing code

Example: A user registration system needs to send emails, log events, update analytics, and trigger notifications—all independently.

---

Solution

The Observer Pattern (Pub-Sub model) decouples publishers from subscribers through an event-driven architecture:

• Publisher: Emits events without knowing who listens
• Subscriber: Registers interest in specific events
• Event Bus: Mediates communication
• Loose Coupling: Publishers and subscribers are independent

This enables scalable, maintainable systems where new listeners attach without modifying existing code.

---

Code Implementation

1. Basic Observer Pattern

// Event Emitter Base Class
class EventEmitter {
  constructor() {
    this.events = {};
  }

  on(eventName, callback) {
    if (!this.events[eventName]) {
      this.events[eventName] = [];
    }
    this.events[eventName].push(callback);

    // Return unsubscribe function
    return () => this.off(eventName, callback);
  }

  off(eventName, callback) {
    if (!this.events[eventName]) return;
    this.events[eventName] = this.events[eventName].filter(
      cb => cb !== callback
    );
  }

  emit(eventName, data) {
    if (!this.events[eventName]) return;
    this.events[eventName].forEach(callback => callback(data));
  }

  once(eventName, callback) {
    const wrapper = (data) => {
      callback(data);
      this.off(eventName, wrapper);
    };
    this.on(eventName, wrapper);
  }
}

// Usage
const userEvents = new EventEmitter();

userEvents.on('user:registered', (user) => {
  console.log(`📧 Sending email to ${user.email}`);
});

userEvents.on('user:registered', (user) => {
  console.log(`📊 Logging user registration: ${user.id}`);
});

userEvents.emit('user:registered', { id: 1, email: 'john@example.com' });

2. Advanced Pub-Sub System

class PubSubBus {
  constructor() {
    this.subscribers = new Map();
    this.history = [];
    this.middleware = [];
  }

  subscribe(topic, handler, options = {}) {
    const { priority = 0, once = false } = options;

    if (!this.subscribers.has(topic)) {
      this.subscribers.set(topic, []);
    }

    const subscription = {
      handler,
      priority,
      once,
      id: Math.random().toString(36).substr(2, 9)
    };

    this.subscribers.get(topic).push(subscription);

    // Sort by priority (higher first)
    this.subscribers.get(topic).sort((a, b) => b.priority - a.priority);

    // Return unsubscribe function
    return () => this.unsubscribe(topic, subscription.id);
  }

  unsubscribe(topic, subscriptionId) {
    if (!this.subscribers.has(topic)) return;

    const subs = this.subscribers.get(topic);
    const index = subs.findIndex(s => s.id === subscriptionId);

    if (index > -1) {
      subs.splice(index, 1);
    }
  }

  use(middlewareFn) {
    this.middleware.push(middlewareFn);
  }

  async publish(topic, data) {
    const event = { topic, data, timestamp: Date.now() };

    // Execute middleware
    for (const mw of this.middleware) {
      await mw(event);
    }

    // Store in history
    this.history.push(event);

    // Execute subscribers
    if (!this.subscribers.has(topic)) return;

    const subs = this.subscribers.get(topic);
    const toRemove = [];

    for (const subscription of subs) {
      try {
        await subscription.handler(data);

        if (subscription.once) {
          toRemove.push(subscription.id);
        }
      } catch (error) {
        console.error(`Error in subscriber for ${topic}:`, error);
      }
    }

    // Remove one-time subscribers
    toRemove.forEach(id => this.unsubscribe(topic, id));
  }

  getHistory(topic) {
    return this.history.filter(e => e.topic === topic);
  }
}

// Usage
const bus = new PubSubBus();

// Middleware for logging
bus.use((event) => {
  console.log(`[${event.timestamp}] Event: ${event.topic}`);
});

// Subscribers
bus.subscribe('order:created', async (order) => {
  console.log(`💳 Processing payment for order ${order.id}`);
  await new Promise(r => setTimeout(r, 100));
}, { priority: 10 });

bus.subscribe('order:created', (order) => {
  console.log(`📦 Preparing shipment for order ${order.id}`);
}, { priority: 5 });

bus.subscribe('order:created', (order) => {
  console.log(`📧 Sending confirmation email`);
}, { priority: 1 });

// Publish event
bus.publish('order:created', { id: 'ORD-001', amount: 99.99 });

3. Real-World Example: User Registration System

class UserService extends EventEmitter {
  constructor() {
    super();
    this.users = [];
  }

  registerUser(userData) {
    const user = {
      id: Date.now(),
      ...userData,
      createdAt: new Date()
    };

    this.users.push(user);

    // Emit event
    this.emit('user:registered', user);

    return user;
  }

  updateUser(userId, updates) {
    const user = this.users.find(u => u.id === userId);
    if (!user) throw new Error('User not found');

    Object.assign(user, updates);
    this.emit('user:updated', user);

    return user;
  }
}

// Services that listen to events
class EmailService {
  constructor(userService) {
    userService.on('user:registered', (user) => {
      this.sendWelcomeEmail(user);
    });

    userService.on('user:updated', (user) => {
      this.sendUpdateConfirmation(user);
    });
  }

  sendWelcomeEmail(user) {
    console.log(`✉️  Welcome email sent to ${user.email}`);
  }

  sendUpdateConfirmation(user) {
    console.log(`✉️  Update confirmation sent to ${user.email}`);
  }
}

class AnalyticsService {
  constructor(userService) {
    userService.on('user:registered', (user) => {
      this.trackSignup(user);
    });
  }

  trackSignup(user) {
    console.log(`📊 Tracked signup: ${user.email}`);
  }
}

class NotificationService {
  constructor(userService) {
    userService.on('user:registered', (user) => {
      this.notifyAdmins(user);
    });
  }

  notifyAdmins(user) {
    console.log(`🔔 Admin notified of new user: ${user.email}`);
  }
}

// Setup
const userService = new UserService();
new EmailService(userService);
new AnalyticsService(userService);
new NotificationService(userService);

// Usage
userService.registerUser({
  email: 'alice@example.com',
  name: 'Alice'
});

4. TypeScript Implementation

interface Subscriber<T> {
  (data: T): void | Promise<void>;
}

interface SubscriptionOptions {
  priority?: number;
  once?: boolean;
}

class TypedEventEmitter<Events extends Record<string, any>> {
  private listeners: Map<keyof Events, Set<Subscriber<any>>> = new Map();

  on<K extends keyof Events>(
    event: K,
    listener: Subscriber<Events[K]>,
    options?: SubscriptionOptions
  ): () => void {
    if (!this.listeners.has(event)) {
      this.listeners.set(event, new Set());
    }

    const set = this.listeners.get(event)!;
    set.add(listener);

    return () => {
      set.delete(listener);
    };
  }

  emit<K extends keyof Events>(event: K, data: Events[K]): void {
    const set = this.listeners.get(event);
    if (!set) return;

    set.forEach(listener => {
      try {
        listener(data);
      } catch (error) {
        console.error(`Error in listener for ${String(event)}:`, error);
      }
    });
  }
}

// Type-safe usage
interface UserEvents {
  'user:created': { id: number; email: string };
  'user:deleted': { id: number };
}

const emitter = new TypedEventEmitter<UserEvents>();

emitter.on('user:created', (user) => {
  console.log(user.email); // Type-safe!
});

emitter.emit('user:created', { id: 1, email: 'test@example.com' });

---

Tips & Best Practices

1. Memory Leaks Prevention

// ❌ Bad: Listener never removed
component.on('event', handler);

// ✅ Good: Store unsubscribe function
const unsubscribe = component.on('event', handler);
component.onDestroy(() => unsubscribe());

2. Error Handling

// Wrap handlers in try-catch to prevent cascade failures
emit(eventName, data) {
  this.events[eventName]?.forEach(callback => {
    try {
      callback(data);
    } catch (error) {
      console.error(`Handler error for ${eventName}:`, error);
    }
  });
}

3. Async Operations

// Use async/await for sequential execution
async publish(topic, data) {
  for (const handler of this.handlers[topic] || []) {
    await handler(data); // Wait for completion
  }
}

4. Event Naming Convention

// Use domain:action format
'user:registered'
'order:shipped'
'payment:failed'
'notification:sent'

5. Debugging

// Add event logging middleware
bus.use((event) => {
  console.log(`📤 ${event.topic}:`, event.data);
});

6. Performance Optimization

// Debounce high-frequency events
const debounce = (fn, delay) => {
  let timeout;
  return (...args) => {
    clearTimeout(timeout);
    timeout = setTimeout(() => fn(...args), delay);
  };
};

emitter.on('scroll', debounce(handler, 300));

---

Summary

The Observer Pattern provides:

• ✅ Loose Coupling: Components don't reference each other
• ✅ Scalability: Add listeners without modifying publishers
• ✅ Maintainability: Clear separation of concerns
• ✅ Testability: Mock events easily
• ✅ Flexibility: Dynamic subscription management

Perfect for event-driven architectures, real-time applications, and complex state management.