Event Bus | IMNS

The event bus is a cornerstone of DurableWS. It enables a pub/sub (publish-subscribe) pattern that drives the library’s flexibility and extensibility. It transforms WebSocket communication into an event-driven architecture, allowing clear and decoupled interactions between different parts of the client.

This article will explore the theory behind the pub/sub pattern, its implementation in DurableWS, and the practical benefits it provides for WebSocket-based applications.

What is the Pub/Sub Pattern?

The pub/sub pattern is a messaging paradigm where:

Publishers emit events or messages without knowing who will receive them.
Subscribers listen for specific events and act upon them when emitted.

This design pattern fosters loose coupling by allowing different components of an application to communicate without being directly dependent on each other.

Why Use Pub/Sub with WebSockets?

WebSockets are inherently event-driven, making the pub/sub pattern a natural fit. For example:

Connection states (open, close, error) can trigger specific behaviors.
Incoming messages can be dispatched to appropriate handlers without hardcoding logic.
Custom behaviors like analytics tracking or logging can be layered without modifying the core logic.

By integrating the pub/sub pattern, we achieve modularity and scalability while remaining intuitive to extend.

Designing the Event Bus

Our event bus is a concrete implementation of the pub/sub pattern. It provides methods to:

Subscribe to events (on method).
Unsubscribe from events (off method).
Publish events (emit method).

Here is the interface definition:

/**
 * Events emitted by the WebSocket client.
 */
export type SocketEvent = "connecting" | "connected" | "close" | "retry" | "disconnect";

/**
 * Event bus interface for managing event subscriptions and emissions.
 */
export interface EventBus {
    /**
     * Subscribes to an event.
     * @param eventName - The name of the event.
     * @param handler - The function to handle the event.
     */
    on<T = unknown>(eventName: string, handler: (payload: T) => void): void;

    /**
     * Unsubscribes from an event.
     * @param eventName - The name of the event.
     * @param handler - The function to remove from the event.
     */
    off<T = unknown>(eventName: string, handler: (payload: T) => void): void;

    /**
     * Emits an event.
     * @param eventName - The name of the event.
     * @param payload - The payload to send with the event.
     */
    emit<T = unknown>(eventName: string, payload?: T): void;
}

Implementing the Event Bus

import type { EventBus } from "@/types";

export function defineEventBus(): EventBus {
    const listeners = new Map<string, Array<(payload: unknown) => void>>();

    function on<T = unknown>(eventName: string, handler: (payload: T) => void): void {
        const handlers = listeners.get(eventName) ?? [];
        handlers.push(handler as (payload: unknown) => void);
        listeners.set(eventName, handlers);
    }

    function off<T = unknown>(eventName: string, handler: (payload: T) => void) {
        const handlers = listeners.get(eventName);
        if (!handlers) return;
        listeners.set(
            eventName,
            handlers.filter((h) => h !== handler)
        );
    }

    function emit<T = unknown>(eventName: string, payload: T) {
        const handlers = listeners.get(eventName);
        handlers?.forEach((fn) => fn(payload));
    }

    return { on, off, emit };
}

The implementation uses a Map to store event listeners, which ensures fast retrieval of handlers for any event. This design minimizes iteration overhead, even as the number of events and listeners grows. For applications handling hundreds of concurrent WebSocket connections, this efficiency ensures low latency and responsiveness.

Using the Event Bus

Although our library uses the event bus internally, it could be helpful in other projects. Here is how it could be used in a standalone project:

const bus = defineEventBus();

// Subscribe to an event
bus.on("connected", () => {
    console.log("WebSocket connected");
});

// Emit an event
bus.emit("connected");

// Unsubscribe from an event
const handler = (payload: { reason: string }) => {
    console.log("Connection closed:", payload.reason);
};
bus.on("close", handler);
bus.off("close", handler);

By decoupling event management from the WebSocket client, you can add or modify behaviors without touching core logic. For example:

Retry Logic: Emit a custom reconnect-attempt event during a retry sequence.
Monitoring: Add a subscriber to log connection state changes for analytics purposes.

We’re eventualy going to add these features to the library, but they serve as a good example of how the event bus can be used.

You could also easily extend the library to include new features like dynamic subscriptions, which would let you manage events based on context, such as joining or leaving WebSocket rooms.

function subscribeToRoom(roomId: string) {
    bus.on(`room:${roomId}:message`, (message) => {
        console.log(`Message in room ${roomId}:`, message);
    });
}

function unsubscribeFromRoom(roomId: string) {
    bus.off(`room:${roomId}:message`, handler);
}

Integrating with the Client

Using the Event Bus for Native WebSocket Events

We’re oging to use the Event Bus to wrap the native WebSocket events (onopen, onmessage, etc.) in a higher-level API. This abstraction makes it easier to subscribe to these events using the on method, without tying your app directly to the native WebSocket API.

Native WebSocket Events: Events like onopen, onmessage, and onerror are directly mapped to eventBus events. For example:

ws.onopen = () => {
    eventBus.emit("connect");
};

ws.onmessage = (event) => {
    eventBus.emit("message", event);
};

This means whenever the WebSocket connection opens, the connect event will be emitted via the eventBus.

The Client’s `on` Method

The on method allows subscribing to specific events emitted by the eventBus. The signature:

function on<T = unknown>(eventName: string, handler: (payload: T) => void) {
    eventBus.on<T>(eventName, handler);
    return () => eventBus.off<T>(eventName, handler);
}

Key Features:

Type Safety:
The generic <T> ensures the handler is strongly typed. For example:

client.on<{ foo: string }>("message", (payload) => {
    console.log(payload.foo);
});

Unsubscribing:
The on method returns a function to remove the handler, allowing easy cleanup:

const off = client.on("connect", () => console.log("Connected!"));
// Later...
off(); // Unsubscribe

By wrapping the Event Bus on method in the client, we can provide a more intuitive API for subscribing to events. This makes it easier to manage event subscriptions and cleanup and allows us to pass the unsubscribe function around if needed.

The Client’s `emit` Method

The emit method provides an abstraction over the native WebSocket send method:

function emit(data: unknown) {
    ws?.send(JSON.stringify(data));
}

How It Works:

The data object is serialized to JSON before being sent over the WebSocket.
You can build higher-level abstractions around this method to handle custom protocols, e.g., adding metadata like event names:
```
function emit(event: string, data: unknown) {
    ws?.send(JSON.stringify({ event, data }));
}
```

This could then be consumed by the server and other connected clients.

Let’s see an example

Finally the cool part.Here’s how you might use the client:

const client = defineClient({ url: "wss://example.com/socket" });

client.on("connect", () => {
    console.log("WebSocket connected!");
});

client.on("message", (msg) => {
    console.log("Received message:", msg);
});

// Send a message
client.emit({ type: "greet", payload: "Hello, server!" });

// Later, clean up the "connect" listener
const offConnect = client.on("connect", () => console.log("Reconnected"));
offConnect();

Advantages of the Event Bus

Decoupled Logic:
Instead of coupling WebSocket events to specific handlers in the client, you emit them to the eventBus. This decouples concerns and allows multiple listeners for the same event.
Custom Event Handling:
You can define new events unrelated to the raw WebSocket (e.g., “reconnect-attempt” or “heartbeat”).
Enhanced Testing:
The eventBus can be mocked or spied on during tests, allowing you to test event-driven behavior without requiring an actual WebSocket connection.

The event bus is a core part of DurableWS’s design. It implements the pub/sub pattern to enable event-driven WebSocket communication. The bus provides the flexibility and scalability needed to build robust real-time applications while keeping the codebase modular and maintainable.

The following article will explore the store pattern, which builds on the event bus to effectively manage states and actions.