TypeScript Pub/Sub Tutorial: A Simple Guide [2024]

The Ultimate TypeScript Pub/Sub Tutorial [2024]

Ever found yourself in a tangled mess of direct function calls, where one component has to know way too much about another? UserService calls NotificationService, which calls LoggingService, which calls AnalyticsService... It's a house of cards. This tight coupling is a recipe for brittle, hard-to-maintain code. One small change in a low-level service can cascade into a series of breaking changes upstream. I've been there, and the refactoring headaches are real.

The elegant solution to this architectural nightmare is a design pattern that's as powerful as it is simple: the Publish-Subscribe, or "Pub/Sub," pattern.

In this comprehensive guide, you will not only learn what the Pub/Sub pattern is in the simplest terms, but you will also build a fully functional, type-safe TypeScript Pub/Sub system from scratch. We'll move from basic theory to a practical, real-world example, giving you the confidence to use this powerful pattern in your own projects. This post is perfect for TypeScript Pub/Sub for beginners and developers looking to understand the fundamentals of event-driven architecture.

By the end of this article, you'll have a deep, practical understanding of one of the most crucial patterns in modern software development. Let's untangle that code.

What is the Pub/Sub Pattern? A Simple Explanation

Before we dive into any TypeScript, let's understand the core concept. Forget about code for a moment.

Imagine you're subscribing to a YouTube channel.

You (the Subscriber) tell YouTube you're interested in a specific channel (the Topic). When the creator (the Publisher) uploads a new video (the Message), YouTube's platform (the Message Broker) automatically notifies you.

The creator doesn't need to know who you are, your email address, or how many other people are subscribed. They just upload their video to the platform. Likewise, you don't need to know the creator's personal upload schedule; you trust the platform to notify you when something new is available. They are completely decoupled.

That's the Pub/Sub pattern in a nutshell. It's a messaging pattern where senders of messages, called publishers, do not program the messages to be sent directly to specific receivers, called subscribers. Instead, the publisher "publishes" messages to a channel or topic, without any knowledge of what subscribers, if any, there may be.

The Core Components

Let's break down this analogy into its technical components:

Publisher: The component that sends messages. It has no knowledge of the subscribers. In our analogy, this is the YouTube creator.
Subscriber: The component that receives messages. It has no knowledge of the publisher. It simply expresses interest in one or more topics. This is you, the viewer.
Topic/Channel: The medium through which messages are sent. Publishers send messages to a topic, and subscribers listen to that topic. This is the YouTube channel itself.
Message: The actual data being transmitted from the publisher to the subscribers. This is the new video notification.
Message Broker (or Event Bus): The central hub that manages subscriptions and routes messages from publishers to the correct subscribers. This is the entire YouTube platform in our analogy.

A Visual Guide

A diagram often makes this clearer than words ever could.

+-----------+                        +-----------------+                        +-------------+
|           | --(1) Publish(Topic A, Msg)--> |                 | --(3) Forward(Msg)--> |             |
| Publisher |                        |  Message Broker |                        | Subscriber A|
|           |                        |                 |                        |             |
+-----------+                        |   (Handles      |                        +-------------+
                                     |    Topics)      |
                                     |                 |                        +-------------+
                                     |                 | --(3) Forward(Msg)--> |             |
                                     +-------+---------+                        | Subscriber B|
                                             |                                  |             |
                                             |                                  +-------------+
                                (2) Subscriber A & B
                                  are interested in
                                       Topic A

The Flow:

The Publisher sends a message to the Message Broker on a specific topic (e.g., user:created). It has no idea who's listening.
Subscribers A and B have previously told the Message Broker they are interested in user:created.
The Message Broker sees the new message on user:created and immediately forwards it to all registered subscribers, A and B.

The key takeaway is the central broker. It's the intermediary that allows the Publisher and Subscribers to be completely ignorant of each other's existence.

Okay, the concept is simple enough. But why should you introduce this extra layer of abstraction into your application? The benefits are massive, especially as your application grows in complexity.

Loose Coupling: This is the big one. Your services no longer need direct references to each other. A UserService doesn't need to import and call methods on a NotificationService. This means you can change, replace, or remove the NotificationService without ever touching the UserService. This drastically reduces the ripple effect of changes.
Improved Scalability: Because components are independent, you can scale them independently. If your notification system is under heavy load, you can add more NotificationService instances (subscribers) without affecting the publishers. The system as a whole becomes more resilient and easier to manage.
Asynchronous Communication: Publishers can fire off an event and immediately move on to their next task without waiting for the subscribers to complete their work. This non-blocking behavior is essential for building responsive, high-performance applications, particularly in Node.js environments.
Separation of Concerns: Each service can focus solely on its core responsibility. The UserService is responsible for user management. The NotificationService is responsible for sending notifications. The AnalyticsService handles analytics. None of them need to know the implementation details of the others. This leads to cleaner, more modular, and easier-to-test code.

Before and After: A Tangible Example

Let's see a decoupling services with Pub/Sub TypeScript example.

Before: The Tightly-Coupled Nightmare

Imagine an e-commerce checkout service.

typescript
// Tightly coupled services
class NotificationService {
  sendOrderConfirmation(orderId: string) {
    console.log(`Sending confirmation email for order ${orderId}...`);
  }
}

class InventoryService {
  decrementStock(productId: string, quantity: number) {
    console.log(`Decrementing stock for product ${productId} by ${quantity}...`);
  }
}

class OrderService {
  private notificationService: NotificationService;
  private inventoryService: InventoryService;

  constructor() {
    // Direct dependencies!
    this.notificationService = new NotificationService();
    this.inventoryService = new InventoryService();
  }

  placeOrder(order: { orderId: string; productId: string; quantity: number }) {
    console.log(`Placing order ${order.orderId}...`);

    // Direct method calls
    this.notificationService.sendOrderConfirmation(order.orderId);
    this.inventoryService.decrementStock(order.productId, order.quantity);

    console.log('Order placed successfully.');
  }
}

const orderService = new OrderService();
orderService.placeOrder({ orderId: '123', productId: 'abc', quantity: 2 });

What's wrong with this?

The OrderService is now responsible for knowing about and instantiating both the NotificationService and InventoryService.
What if we want to add an AnalyticsService? We have to modify the OrderService class.
What if sendOrderConfirmation becomes asynchronous? The placeOrder method has to be updated to handle that.
Testing OrderService is a pain because we have to mock two other classes.

After: The Decoupled Dream (Conceptual)

Now, let's refactor this using the Pub/Sub pattern. We won't write the full implementation yet, but let's see the design.

+--------------+
| OrderService |
+--------------+
      |
      | (1) Publishes "order:placed" event with order data
      v
+----------------+
|  Message Bus   |
+----------------+
      |
      +-----> (2) Notifies interested subscribers
      |
      v
+---------------------+      +--------------------+
| NotificationService |      |  InventoryService  |
+---------------------+      +--------------------+
      |                              |
      | (3) Subscribes to            | (3) Subscribes to
      |   "order:placed"             |   "order:placed"
      |   and sends email            |   and decrements stock

In this new world, the OrderService's placeOrder method would look something like this:

typescript
// Conceptual code
class OrderService {
  constructor(private eventBus: PubSub) {}

  placeOrder(order: OrderPayload) {
    console.log(`Placing order ${order.orderId}...`);

    // Just publish an event. That's it!
    this.eventBus.publish('order:placed', order);

    console.log('Order placed successfully.');
  }
}

Look how clean that is! The OrderService has one job: place the order and announce that it happened. It has no idea what happens next. We can now add 10 more services that react to an order being placed (analytics, shipping, fraud detection) without ever touching a single line of code in OrderService. That's the power of decoupling.

How to Implement Pub/Sub from Scratch in TypeScript

Before we reach for external libraries like Redis or RabbitMQ, let's build our own simple Pub/Sub implementation in TypeScript. This exercise is incredibly valuable because it demystifies the pattern and solidifies your understanding of the core mechanics. We're going to build a generic, type-safe PubSub class that can be used in any TypeScript project.

Step 1: Defining Type-Safe Interfaces

The magic of using TypeScript is, well, the types! We want to ensure that when we publish an event for a specific topic, the data payload matches what the subscribers for that topic expect. If we publish a user:created event, the payload should be a User object, not a Product object. Generics will be our best friend here.

Let's start by defining our core types:

typescript
// Core types for our Pub/Sub system
type Callback<T = any> = (data: T) => void | Promise<void>;

interface Subscription {
  id: string;
  callback: Callback;
}

interface EventMap {
  [key: string]: any;
}

// Type-safe event definitions
interface AppEvents extends EventMap {
  'user:created': { id: string; email: string; name: string };
  'user:updated': { id: string; changes: Partial<{ email: string; name: string }> };
  'order:placed': { orderId: string; userId: string; productId: string; quantity: number; total: number };
  'order:cancelled': { orderId: string; reason: string };
  'notification:sent': { type: 'email' | 'sms'; recipient: string; messageId: string };
}

This AppEvents interface is our event registry. It's a TypeScript mapped type that defines all the events our application can emit, along with their exact payload structure. This ensures compile-time safety—you can't accidentally publish the wrong data type for an event.

Step 2: Building the Core PubSub Class

Now, let's implement our type-safe PubSub class:

typescript
class PubSub<TEventMap extends EventMap = EventMap> {
  private subscriptions: Map<keyof TEventMap, Subscription[]> = new Map();
  private nextId: number = 1;

  /**
   * Subscribe to an event
   */
  subscribe<K extends keyof TEventMap>(
    event: K,
    callback: Callback<TEventMap[K]>
  ): string {
    const subscriptionId = `sub_${this.nextId++}`;
    
    if (!this.subscriptions.has(event)) {
      this.subscriptions.set(event, []);
    }
    
    const subscription: Subscription = {
      id: subscriptionId,
      callback: callback as Callback
    };
    
    this.subscriptions.get(event)!.push(subscription);
    
    console.log(`✅ Subscribed to "${String(event)}" with ID: ${subscriptionId}`);
    return subscriptionId;
  }

  /**
   * Unsubscribe from an event
   */
  unsubscribe(subscriptionId: string): boolean {
    for (const [event, subs] of this.subscriptions.entries()) {
      const index = subs.findIndex(sub => sub.id === subscriptionId);
      if (index !== -1) {
        subs.splice(index, 1);
        console.log(`❌ Unsubscribed from "${String(event)}" (ID: ${subscriptionId})`);
        
        // Clean up empty subscription arrays
        if (subs.length === 0) {
          this.subscriptions.delete(event);
        }
        return true;
      }
    }
    return false;
  }

  /**
   * Publish an event to all subscribers
   */
  async publish<K extends keyof TEventMap>(
    event: K,
    data: TEventMap[K]
  ): Promise<void> {
    const subscribers = this.subscriptions.get(event) || [];
    
    if (subscribers.length === 0) {
      console.log(`⚠️  No subscribers for event "${String(event)}"`);
      return;
    }

    console.log(`📢 Publishing "${String(event)}" to ${subscribers.length} subscriber(s)`);
    
    // Execute all callbacks concurrently
    const promises = subscribers.map(async (subscription) => {
      try {
        await subscription.callback(data);
      } catch (error) {
        console.error(`❌ Error in subscriber ${subscription.id}:`, error);
        // Continue executing other subscribers even if one fails
      }
    });

    await Promise.allSettled(promises);
  }

  /**
   * Publish an event synchronously (fire and forget)
   */
  publishSync<K extends keyof TEventMap>(
    event: K,
    data: TEventMap[K]
  ): void {
    const subscribers = this.subscriptions.get(event) || [];
    
    if (subscribers.length === 0) {
      console.log(`⚠️  No subscribers for event "${String(event)}"`);
      return;
    }

    console.log(`📢 Publishing "${String(event)}" synchronously to ${subscribers.length} subscriber(s)`);
    
    subscribers.forEach((subscription) => {
      try {
        subscription.callback(data);
      } catch (error) {
        console.error(`❌ Error in subscriber ${subscription.id}:`, error);
        // Continue executing other subscribers even if one fails
      }
    });
  }

  /**
   * Get the number of subscribers for an event
   */
  getSubscriberCount<K extends keyof TEventMap>(event: K): number {
    return this.subscriptions.get(event)?.length || 0;
  }

  /**
   * Clear all subscriptions for an event or all events
   */
  clear<K extends keyof TEventMap>(event?: K): void {
    if (event) {
      this.subscriptions.delete(event);
      console.log(`🧹 Cleared all subscriptions for "${String(event)}"`);
    } else {
      this.subscriptions.clear();
      console.log(`🧹 Cleared all subscriptions`);
    }
  }

  /**
   * Get debug information about the current state
   */
  debug(): void {
    console.log('📊 PubSub Debug Info:');
    console.log(`Total events: ${this.subscriptions.size}`);
    
    for (const [event, subs] of this.subscriptions.entries()) {
      console.log(`  ${String(event)}: ${subs.length} subscriber(s)`);
    }
  }
}

Step 3: Creating a Global Event Bus

For most applications, you'll want a single, global event bus that all your services can use:

typescript
// Create a global, typed event bus
export const eventBus = new PubSub<AppEvents>();

// Convenience functions for better ergonomics
export const subscribe = eventBus.subscribe.bind(eventBus);
export const unsubscribe = eventBus.unsubscribe.bind(eventBus);
export const publish = eventBus.publish.bind(eventBus);
export const publishSync = eventBus.publishSync.bind(eventBus);

Real-World Example: Building an E-Commerce System

Let's put our PubSub system to work with a practical e-commerce example. We'll build several services that communicate through events instead of direct method calls.

Step 4: Defining Our Services

typescript
// User data types
interface User {
  id: string;
  email: string;
  name: string;
}

interface Order {
  orderId: string;
  userId: string;
  productId: string;
  quantity: number;
  total: number;
}

// Service implementations
class UserService {
  private users: Map<string, User> = new Map();

  constructor() {
    // Subscribe to relevant events
    subscribe('user:updated', this.handleUserUpdate.bind(this));
  }

  async createUser(userData: Omit<User, 'id'>): Promise<User> {
    const user: User = {
      id: `user_${Date.now()}`,
      ...userData
    };

    this.users.set(user.id, user);
    console.log(`👤 Created user: ${user.name} (${user.email})`);

    // Publish event - other services can react to this
    await publish('user:created', user);

    return user;
  }

  async updateUser(id: string, changes: Partial<Pick<User, 'email' | 'name'>>): Promise<void> {
    const user = this.users.get(id);
    if (!user) {
      throw new Error(`User ${id} not found`);
    }

    Object.assign(user, changes);
    console.log(`👤 Updated user: ${user.name}`);

    // Publish update event
    await publish('user:updated', { id, changes });
  }

  private async handleUserUpdate(data: AppEvents['user:updated']): Promise<void> {
    console.log(`👤 UserService received user update for ${data.id}`);
    // Handle any internal logic needed when a user is updated
  }

  getUser(id: string): User | undefined {
    return this.users.get(id);
  }
}

class OrderService {
  private orders: Map<string, Order> = new Map();

  constructor() {
    // React to user creation - maybe send welcome email with special offer
    subscribe('user:created', this.handleNewUser.bind(this));
  }

  async placeOrder(orderData: Omit<Order, 'orderId'>): Promise<Order> {
    const order: Order = {
      orderId: `order_${Date.now()}`,
      ...orderData
    };

    this.orders.set(order.orderId, order);
    console.log(`🛒 Placed order: ${order.orderId} for user ${order.userId}`);

    // Publish the order event - this will trigger inventory, notifications, etc.
    await publish('order:placed', order);

    return order;
  }

  async cancelOrder(orderId: string, reason: string): Promise<void> {
    const order = this.orders.get(orderId);
    if (!order) {
      throw new Error(`Order ${orderId} not found`);
    }

    this.orders.delete(orderId);
    console.log(`🚫 Cancelled order: ${orderId}`);

    await publish('order:cancelled', { orderId, reason });
  }

  private async handleNewUser(userData: AppEvents['user:created']): Promise<void> {
    console.log(`🛒 OrderService: Welcome ${userData.name}! Here's a 10% discount code: WELCOME10`);
  }
}

class NotificationService {
  constructor() {
    // Subscribe to multiple events
    subscribe('user:created', this.sendWelcomeEmail.bind(this));
    subscribe('order:placed', this.sendOrderConfirmation.bind(this));
    subscribe('order:cancelled', this.sendCancellationEmail.bind(this));
  }

  private async sendWelcomeEmail(userData: AppEvents['user:created']): Promise<void> {
    console.log(`📧 Sending welcome email to ${userData.email}`);
    
    // Simulate email sending delay
    await new Promise(resolve => setTimeout(resolve, 100));
    
    await publish('notification:sent', {
      type: 'email',
      recipient: userData.email,
      messageId: `welcome_${userData.id}`
    });
  }

  private async sendOrderConfirmation(orderData: AppEvents['order:placed']): Promise<void> {
    console.log(`📧 Sending order confirmation for order ${orderData.orderId}`);
    
    await new Promise(resolve => setTimeout(resolve, 100));
    
    await publish('notification:sent', {
      type: 'email',
      recipient: `user_${orderData.userId}@example.com`,
      messageId: `order_${orderData.orderId}`
    });
  }

  private async sendCancellationEmail(data: AppEvents['order:cancelled']): Promise<void> {
    console.log(`📧 Sending cancellation email for order ${data.orderId}`);
    
    await publish('notification:sent', {
      type: 'email',
      recipient: 'customer@example.com',
      messageId: `cancel_${data.orderId}`
    });
  }
}

class InventoryService {
  private inventory: Map<string, number> = new Map();

  constructor() {
    // Initialize some test inventory
    this.inventory.set('product_1', 100);
    this.inventory.set('product_2', 50);

    // Subscribe to order events
    subscribe('order:placed', this.decrementStock.bind(this));
    subscribe('order:cancelled', this.incrementStock.bind(this));
  }

  private async decrementStock(orderData: AppEvents['order:placed']): Promise<void> {
    const currentStock = this.inventory.get(orderData.productId) || 0;
    const newStock = Math.max(0, currentStock - orderData.quantity);
    
    this.inventory.set(orderData.productId, newStock);
    console.log(`📦 Decremented stock for ${orderData.productId}: ${currentStock} → ${newStock}`);
    
    if (newStock <= 5) {
      console.log(`⚠️  Low stock alert for product ${orderData.productId}: only ${newStock} left!`);
    }
  }

  private async incrementStock(data: AppEvents['order:cancelled']): Promise<void> {
    // In a real app, you'd need to look up the original order to know the quantity
    // For this demo, we'll just add back a fixed amount
    const productId = 'product_1'; // Simplified for demo
    const currentStock = this.inventory.get(productId) || 0;
    const newStock = currentStock + 1;
    
    this.inventory.set(productId, newStock);
    console.log(`📦 Incremented stock for ${productId}: ${currentStock} → ${newStock}`);
  }

  getStock(productId: string): number {
    return this.inventory.get(productId) || 0;
  }
}

class AnalyticsService {
  private events: Array<{ event: string; timestamp: Date; data: any }> = [];

  constructor() {
    // Subscribe to all events for analytics
    subscribe('user:created', (data) => this.trackEvent('user:created', data));
    subscribe('order:placed', (data) => this.trackEvent('order:placed', data));
    subscribe('order:cancelled', (data) => this.trackEvent('order:cancelled', data));
    subscribe('notification:sent', (data) => this.trackEvent('notification:sent', data));
  }

  private trackEvent(eventName: string, data: any): void {
    this.events.push({
      event: eventName,
      timestamp: new Date(),
      data
    });
    console.log(`📊 Analytics: Tracked ${eventName}`);
  }

  getEventCount(): number {
    return this.events.length;
  }

  getEventsByType(eventType: string): any[] {
    return this.events.filter(e => e.event === eventType).map(e => e.data);
  }
}

Step 5: Putting It All Together

Let's create an example that demonstrates our entire system working together:

typescript
async function runECommerceDemo() {
  console.log('🚀 Starting E-Commerce Demo\n');

  // Initialize all services
  const userService = new UserService();
  const orderService = new OrderService();
  const notificationService = new NotificationService();
  const inventoryService = new InventoryService();
  const analyticsService = new AnalyticsService();

  console.log('📊 Initial state:');
  eventBus.debug();
  console.log(`📦 Product 1 stock: ${inventoryService.getStock('product_1')}`);
  console.log('');

  try {
    // 1. Create a new user
    console.log('🔄 Step 1: Creating new user...');
    const user = await userService.createUser({
      name: 'Alice Johnson',
      email: 'alice@example.com'
    });
    console.log('');

    // 2. Place an order
    console.log('🔄 Step 2: Placing order...');
    const order = await orderService.placeOrder({
      userId: user.id,
      productId: 'product_1',
      quantity: 3,
      total: 299.99
    });
    console.log('');

    // 3. Update user information
    console.log('🔄 Step 3: Updating user...');
    await userService.updateUser(user.id, { name: 'Alice Smith' });
    console.log('');

    // 4. Cancel an order
    console.log('🔄 Step 4: Cancelling order...');
    await orderService.cancelOrder(order.orderId, 'Customer changed mind');
    console.log('');

    // 5. Show final state
    console.log('📊 Final state:');
    console.log(`📦 Product 1 stock: ${inventoryService.getStock('product_1')}`);
    console.log(`📈 Total events tracked: ${analyticsService.getEventCount()}`);
    console.log(`👥 Users created: ${analyticsService.getEventsByType('user:created').length}`);
    console.log(`🛒 Orders placed: ${analyticsService.getEventsByType('order:placed').length}`);

  } catch (error) {
    console.error('❌ Demo failed:', error);
  }
}

// Run the demo
runECommerceDemo().catch(console.error);

Advanced Features and Best Practices

Error Handling and Resilience

Our basic implementation handles errors by logging them and continuing with other subscribers. For production systems, you might want more sophisticated error handling:

typescript
interface SubscriptionOptions {
  maxRetries?: number;
  retryDelay?: number;
  errorHandler?: (error: Error, data: any) => void;
}

class RobustPubSub<TEventMap extends EventMap = EventMap> extends PubSub<TEventMap> {
  async publishWithRetry<K extends keyof TEventMap>(
    event: K,
    data: TEventMap[K],
    options: SubscriptionOptions = {}
  ): Promise<void> {
    const { maxRetries = 3, retryDelay = 1000, errorHandler } = options;
    const subscribers = this.subscriptions.get(event) || [];

    const executeWithRetry = async (subscription: Subscription, attempt: number = 1): Promise<void> => {
      try {
        await subscription.callback(data);
      } catch (error) {
        if (attempt <= maxRetries) {
          console.log(`🔄 Retrying subscription ${subscription.id}, attempt ${attempt}/${maxRetries}`);
          await new Promise(resolve => setTimeout(resolve, retryDelay * attempt));
          return executeWithRetry(subscription, attempt + 1);
        } else {
          if (errorHandler) {
            errorHandler(error as Error, data);
          } else {
            console.error(`❌ Max retries exceeded for subscription ${subscription.id}:`, error);
          }
        }
      }
    };

    const promises = subscribers.map(subscription => executeWithRetry(subscription));
    await Promise.allSettled(promises);
  }
}

Event Middleware and Interceptors

Add middleware to intercept and potentially modify events before they reach subscribers:

typescript
type Middleware<T = any> = (event: string, data: T, next: () => void) => void | Promise<void>;

class MiddlewarePubSub<TEventMap extends EventMap = EventMap> extends PubSub<TEventMap> {
  private middleware: Middleware[] = [];

  use(middleware: Middleware): void {
    this.middleware.push(middleware);
  }

  async publish<K extends keyof TEventMap>(
    event: K,
    data: TEventMap[K]
  ): Promise<void> {
    let index = 0;
    
    const next = async (): Promise<void> => {
      if (index < this.middleware.length) {
        const middleware = this.middleware[index++];
        await middleware(String(event), data, next);
      } else {
        // All middleware executed, publish the event
        await super.publish(event, data);
      }
    };

    await next();
  }
}

// Example middleware usage
const eventBusWithMiddleware = new MiddlewarePubSub<AppEvents>();

// Logging middleware
eventBusWithMiddleware.use(async (event, data, next) => {
  console.log(`🔍 Middleware: Processing event ${event}`, data);
  await next();
  console.log(`✅ Middleware: Completed event ${event}`);
});

// Validation middleware
eventBusWithMiddleware.use(async (event, data, next) => {
  if (event === 'user:created' && !data.email.includes('@')) {
    throw new Error('Invalid email address');
  }
  await next();
});

Event Namespacing and Wildcards

For larger applications, you might want event namespacing and wildcard subscriptions:

typescript
class NamespacedPubSub<TEventMap extends EventMap = EventMap> extends PubSub<TEventMap> {
  subscribeToNamespace(namespace: string, callback: Callback): string {
    // Subscribe to all events that start with the namespace
    const subscriptionId = `ns_${this.nextId++}`;
    
    // Store namespace subscriptions separately
    // Implementation would check event names against namespace patterns
    
    return subscriptionId;
  }

  subscribeWithWildcard(pattern: string, callback: Callback): string {
    // Support patterns like 'user:*' or 'order:*'
    // Implementation would use regex or glob matching
    
    return `wildcard_${this.nextId++}`;
  }
}

Testing Your Pub/Sub System

Testing event-driven systems requires a different approach than testing traditional function calls. Here's how to test our PubSub system effectively:

typescript
// Test utilities
class TestPubSub<TEventMap extends EventMap = EventMap> extends PubSub<TEventMap> {
  private publishedEvents: Array<{ event: keyof TEventMap; data: any; timestamp: Date }> = [];

  async publish<K extends keyof TEventMap>(
    event: K,
    data: TEventMap[K]
  ): Promise<void> {
    // Record the event for testing
    this.publishedEvents.push({
      event,
      data,
      timestamp: new Date()
    });

    // Call the original publish method
    await super.publish(event, data);
  }

  getPublishedEvents(): typeof this.publishedEvents {
    return [...this.publishedEvents];
  }

  clearPublishedEvents(): void {
    this.publishedEvents = [];
  }

  waitForEvent<K extends keyof TEventMap>(
    event: K,
    timeout: number = 1000
  ): Promise<TEventMap[K]> {
    return new Promise((resolve, reject) => {
      const timeoutId = setTimeout(() => {
        this.unsubscribe(subscriptionId);
        reject(new Error(`Timeout waiting for event ${String(event)}`));
      }, timeout);

      const subscriptionId = this.subscribe(event, (data) => {
        clearTimeout(timeoutId);
        this.unsubscribe(subscriptionId);
        resolve(data);
      });
    });
  }
}

// Example test cases
describe('PubSub System', () => {
  let testEventBus: TestPubSub<AppEvents>;

  beforeEach(() => {
    testEventBus = new TestPubSub<AppEvents>();
  });

  test('should publish and receive events', async () => {
    const receivedData: AppEvents['user:created'][] = [];
    
    testEventBus.subscribe('user:created', (data) => {
      receivedData.push(data);
    });

    const userData = { id: '123', email: 'test@example.com', name: 'Test User' };
    await testEventBus.publish('user:created', userData);

    expect(receivedData).toHaveLength(1);
    expect(receivedData[0]).toEqual(userData);
  });

  test('should handle multiple subscribers', async () => {
    let subscriber1Called = false;
    let subscriber2Called = false;

    testEventBus.subscribe('user:created', () => { subscriber1Called = true; });
    testEventBus.subscribe('user:created', () => { subscriber2Called = true; });

    await testEventBus.publish('user:created', { id: '123', email: 'test@example.com', name: 'Test' });

    expect(subscriber1Called).toBe(true);
    expect(subscriber2Called).toBe(true);
  });

  test('should wait for specific events', async () => {
    setTimeout(() => {
      testEventBus.publish('order:placed', {
        orderId: '123',
        userId: '456',
        productId: 'prod1',
        quantity: 2,
        total: 99.99
      });
    }, 100);

    const eventData = await testEventBus.waitForEvent('order:placed');
    expect(eventData.orderId).toBe('123');
  });

  test('should track published events', async () => {
    await testEventBus.publish('user:created', { id: '1', email: 'test@example.com', name: 'Test' });
    await testEventBus.publish('order:placed', { orderId: '1', userId: '1', productId: 'p1', quantity: 1, total: 10 });

    const events = testEventBus.getPublishedEvents();
    expect(events).toHaveLength(2);
    expect(events[0].event).toBe('user:created');
    expect(events[1].event).toBe('order:placed');
  });
});

Performance Considerations and Optimization

Memory Management

For long-running applications, consider implementing cleanup strategies:

typescript
class OptimizedPubSub<TEventMap extends EventMap = EventMap> extends PubSub<TEventMap> {
  private maxSubscribers: number = 1000;
  private subscriptionMetrics: Map<string, { createdAt: Date; lastUsed: Date }> = new Map();

  subscribe<K extends keyof TEventMap>(
    event: K,
    callback: Callback<TEventMap[K]>
  ): string {
    // Check subscription limits
    const totalSubscriptions = Array.from(this.subscriptions.values())
      .reduce((total, subs) => total + subs.length, 0);

    if (totalSubscriptions >= this.maxSubscribers) {
      throw new Error('Maximum number of subscriptions reached');
    }

    const subscriptionId = super.subscribe(event, callback);
    
    // Track subscription metrics
    this.subscriptionMetrics.set(subscriptionId, {
      createdAt: new Date(),
      lastUsed: new Date()
    });

    return subscriptionId;
  }

  // Clean up old, unused subscriptions
  cleanupOldSubscriptions(maxAge: number = 24 * 60 * 60 * 1000): number {
    const now = new Date();
    let cleanedCount = 0;

    for (const [subscriptionId, metrics] of this.subscriptionMetrics.entries()) {
      if (now.getTime() - metrics.lastUsed.getTime() > maxAge) {
        this.unsubscribe(subscriptionId);
        this.subscriptionMetrics.delete(subscriptionId);
        cleanedCount++;
      }
    }

    return cleanedCount;
  }
}

Batching and Throttling

For high-throughput scenarios, consider batching events:

typescript
class BatchedPubSub<TEventMap extends EventMap = EventMap> extends PubSub<TEventMap> {
  private eventBatch: Array<{ event: keyof TEventMap; data: any }> = [];
  private batchTimeout: NodeJS.Timeout | null = null;
  private batchSize: number = 10;
  private batchDelay: number = 100;

  publishBatched<K extends keyof TEventMap>(
    event: K,
    data: TEventMap[K]
  ): void {
    this.eventBatch.push({ event, data });

    if (this.eventBatch.length >= this.batchSize) {
      this.flushBatch();
    } else if (!this.batchTimeout) {
      this.batchTimeout = setTimeout(() => this.flushBatch(), this.batchDelay);
    }
  }

  private async flushBatch(): Promise<void> {
    if (this.batchTimeout) {
      clearTimeout(this.batchTimeout);
      this.batchTimeout = null;
    }

    const batch = this.eventBatch.splice(0);
    
    // Process all events in the batch
    for (const { event, data } of batch) {
      await this.publish(event as any, data);
    }
  }
}

Integration with External Message Brokers

While our in-memory implementation is perfect for single-process applications, distributed systems need external message brokers. Here's how to adapt our pattern for Redis or RabbitMQ:

Redis Adapter

typescript
import Redis from 'ioredis';

class RedisPubSub<TEventMap extends EventMap = EventMap> {
  private publisher: Redis;
  private subscriber: Redis;
  private callbacks: Map<keyof TEventMap, Callback[]> = new Map();

  constructor(redisUrl: string) {
    this.publisher = new Redis(redisUrl);
    this.subscriber = new Redis(redisUrl);
    
    this.subscriber.on('message', (channel, message) => {
      this.handleMessage(channel, message);
    });
  }

  subscribe<K extends keyof TEventMap>(
    event: K,
    callback: Callback<TEventMap[K]>
  ): string {
    if (!this.callbacks.has(event)) {
      this.callbacks.set(event, []);
      // Subscribe to Redis channel
      this.subscriber.subscribe(String(event));
    }
    
    this.callbacks.get(event)!.push(callback);
    return `redis_sub_${Date.now()}`;
  }

  async publish<K extends keyof TEventMap>(
    event: K,
    data: TEventMap[K]
  ): Promise<void> {
    await this.publisher.publish(String(event), JSON.stringify(data));
  }

  private handleMessage(channel: string, message: string): void {
    const callbacks = this.callbacks.get(channel as keyof TEventMap) || [];
    const data = JSON.parse(message);
    
    callbacks.forEach(callback => {
      try {
        callback(data);
      } catch (error) {
        console.error('Error in Redis subscriber:', error);
      }
    });
  }
}

Conclusion: Mastering Event-Driven Architecture

Congratulations! You've just built a comprehensive, type-safe Pub/Sub system from scratch and learned how to apply it to real-world scenarios. This pattern is more than just a nice-to-have—it's essential for building scalable, maintainable applications.

Key Takeaways

Loose Coupling is King: By decoupling your services through events, you've made your system more flexible, testable, and maintainable. Services can evolve independently without breaking each other.

TypeScript Makes It Better: The type safety we've built ensures that your events are consistent and your data contracts are enforced at compile time. No more runtime surprises from mismatched event payloads.

Scalability Through Design: The Pub/Sub pattern naturally supports horizontal scaling. You can add new subscribers to react to events without modifying existing publishers.

Testing Gets Easier: Event-driven systems are inherently more testable because you can easily mock subscribers and assert on published events.

What's Next?

The Pub/Sub pattern is just the beginning of event-driven architecture. Consider exploring:

Event Sourcing: Store your application state as a sequence of events rather than current state snapshots.

CQRS (Command Query Responsibility Segregation): Separate your read and write models, often combined with event-driven patterns.

Saga Pattern: Manage complex, multi-service transactions using events and compensating actions.

Event Streaming: For high-throughput scenarios, consider platforms like Apache Kafka or AWS Kinesis.

A Final Word

The journey from tightly-coupled services to elegant, event-driven architecture isn't always straightforward, but the benefits are undeniable. You now have the tools and knowledge to build systems that are not just functional, but truly scalable and maintainable.

Start small—refactor one tightly-coupled interaction in your current project using the patterns we've covered. You'll be amazed at how much cleaner your code becomes and how much easier it is to add new features without fear of breaking existing functionality.

The future of software architecture is event-driven, and you're now equipped to be part of that future.

The Ultimate TypeScript Pub/Sub Tutorial

The Ultimate TypeScript Pub/Sub Tutorial [2024]

What is the Pub/Sub Pattern? A Simple Explanation

The Core Components

A Visual Guide

Before and After: A Tangible Example

How to Implement Pub/Sub from Scratch in TypeScript

Step 1: Defining Type-Safe Interfaces

Step 2: Building the Core PubSub Class

Step 3: Creating a Global Event Bus

Real-World Example: Building an E-Commerce System

Step 4: Defining Our Services

Step 5: Putting It All Together

Advanced Features and Best Practices

Error Handling and Resilience

Event Middleware and Interceptors

Event Namespacing and Wildcards

Testing Your Pub/Sub System

Performance Considerations and Optimization

Memory Management

Batching and Throttling

Integration with External Message Brokers

Redis Adapter

Conclusion: Mastering Event-Driven Architecture

Key Takeaways

What's Next?

A Final Word

How to Add AI to Your Existing Web App Without Rewriting It

Why Your RAG Pipeline Is Probably Broken (And How to Fix It)

The Wake-Up Call: How I Learned to Stop Worrying and Secure My Next.js Apps

The Ultimate TypeScript Pub/Sub Tutorial [2024]

What is the Pub/Sub Pattern? A Simple Explanation

The Core Components

A Visual Guide

Why Use the Publish-Subscribe Pattern in TypeScript?

Before and After: A Tangible Example

How to Implement Pub/Sub from Scratch in TypeScript

Step 1: Defining Type-Safe Interfaces

Step 2: Building the Core PubSub Class

Step 3: Creating a Global Event Bus

Real-World Example: Building an E-Commerce System

Step 4: Defining Our Services

Step 5: Putting It All Together

Advanced Features and Best Practices

Error Handling and Resilience

Event Middleware and Interceptors

Event Namespacing and Wildcards

Testing Your Pub/Sub System

Performance Considerations and Optimization

Memory Management

Batching and Throttling

Integration with External Message Brokers

Redis Adapter

Conclusion: Mastering Event-Driven Architecture

Key Takeaways

What's Next?

A Final Word

How to Add AI to Your Existing Web App Without Rewriting It

Why Your RAG Pipeline Is Probably Broken (And How to Fix It)

The Wake-Up Call: How I Learned to Stop Worrying and Secure My Next.js Apps