DEV Community: Red John

Building a Distributed PubSub Class with Cross-Process and Cross-Tab Capabilities

Red John — Sat, 19 Oct 2024 15:11:07 +0000

In modern development, there’s often a need to synchronize messages across multiple execution contexts, be it across processes in Node.js, clustered workers, or even across browser tabs. The PubSub class we’ll build in this post can achieve that goal by using the BroadcastChannel API.

This PubSub class is designed to function:

Across processes when using Bun or Node.js's Cluster module.
Across multiple worker threads using Deno's --parallel feature.
In browser environments, facilitating communication between multiple tabs.

By leveraging the BroadcastChannel API, our PubSub system will enable communication between different contexts seamlessly. We'll also implement asynchronous iteration, so you can consume messages from this system using for await...of syntax in JavaScript.

The `PubSub` Class Design

Our PubSub class will:

Broadcast messages: Share events with any subscriber, across different processes or browser tabs.
Support asynchronous iteration: Using the createEventIterator helper from the previous post to enable iterating over events.
Offer simple subscribe/unsubscribe mechanics: Listeners can subscribe to receive messages and later unsubscribe if no longer interested.
Work seamlessly across different environments: Whether it's browser tabs or Node.js clusters, it provides a uniform API.

The Code

Here’s how you can implement the PubSub class:

type Listener<T> = (data: T) => void;

export class PubSub<T> implements AsyncIterable<T> {
    #listeners = new Set<Listener<T>>();
    #tx: BroadcastChannel;
    #rx: BroadcastChannel;

    [Symbol.asyncIterator](): AsyncIterator<T> {
        return this.iter();
    }

    constructor(name: string) {
        this.#tx = new BroadcastChannel(name);
        this.#rx = new BroadcastChannel(name);

        // Listen for messages on the receive channel
        this.#rx.onmessage = (ev) => {
            this.#broadcast(ev.data);
        };
    }

    #broadcast(data: T) {
        for (const listener of this.#listeners) {
            listener(data);
        }
    }

    close(): void {
        this.#listeners.clear();
        this.#tx.close();
        this.#rx.close();
    }

    broadcast(data: T): void {
        this.#tx.postMessage(data);
    }

    iter(): AsyncIterator<T> {
        return createEventIterator<T>(({ emit }) => this.subscribe(emit));
    }

    once(listener: Listener<T>): () => void {
        const wrapper: Listener<T> = (data) => {
            this.unsubscribe(wrapper);
            listener(data);
        };
        return this.subscribe(wrapper);
    }

    subscribe(listener: Listener<T>): () => void {
        this.#listeners.add(listener);
        return () => this.unsubscribe(listener);
    }

    unsubscribe(listener: Listener<T>): void {
        this.#listeners.delete(listener);
    }
}

Key Components

Broadcast Channels: The class uses two BroadcastChannel instances, one for sending (#tx) and one for receiving (#rx) messages. Each instance is tied to a specific channel name.
Asynchronous Iteration: The class implements the AsyncIterable interface via the [Symbol.asyncIterator]() method, which returns an async iterator. This is where the createEventIterator function, discussed in the previous post, comes into play. It allows you to iterate over incoming events asynchronously.
Subscriptions: You can subscribe multiple listeners to the same broadcast. Each listener will be triggered whenever a message is broadcasted on the channel. You can also register one-time listeners with the once() method.
Cross-context communication: Thanks to the BroadcastChannel API, this works across different execution contexts. For example, in browsers, it works across different tabs, while in Node.js or Deno, it works across different processes or threads.

Usage in Different Environments

1. Node.js / Bun Cluster with Typed Messages

Let's improve the example by using a discriminated union type to represent different message types. This adds a layer of type-safety and clarity to how messages are handled.

type Message = 
  | { type: 'user-connected', userId: number } 
  | { type: 'user-disconnected' } 
  | { type: 'message', content: string, userId: number };

const pubsub = new PubSub<Message>('my-channel');

// Broadcast messages across processes
if (process.isMaster) {
    setInterval(() => {
        pubsub.broadcast({ type: 'user-connected', userId: 1 });
        pubsub.broadcast({ type: 'message', content: 'Hello, world!', userId: 1 });
    }, 1000);
} else {
    // Worker process: Listen for messages
    (async () => {
        for await (const message of pubsub) {
            switch (message.type) {
                case 'user-connected':
                    console.log(`User connected: ${message.userId}`);
                    break;
                case 'user-disconnected':
                    console.log('User disconnected');
                    break;
                case 'message':
                    console.log(`User ${message.userId} says: ${message.content}`);
                    break;
            }
        }
    })();
}

This ensures that every message is type-safe and you can handle different message types using TypeScript’s type system.

2. Deno Workers with `--parallel` and Typed Events

type Message = 
  | { type: 'job-started', jobId: string } 
  | { type: 'job-completed', jobId: string };

const pubsub = new PubSub<Message>('deno-channel');

if (Deno.args[0] === 'worker') {
    // Worker process: Listen for messages
    (async () => {
        for await (const message of pubsub) {
            switch (message.type) {
                case 'job-started':
                    console.log(`Job ${message.jobId} started`);
                    break;
                case 'job-completed':
                    console.log(`Job ${message.jobId} completed`);
                    break;
            }
        }
    })();
} else {
    // Main process: Broadcast messages to workers
    setInterval(() => {
        pubsub.broadcast({ type: 'job-started', jobId: '12345' });
        setTimeout(() => pubsub.broadcast({ type: 'job-completed', jobId: '12345' }), 3000);
    }, 5000);
}

3. Cross-Tab Communication in the Browser

Here’s an example where you can send and receive typed chat messages across multiple browser tabs.

type Message = 
  | { type: 'chat-message', content: string, userId: number } 
  | { type: 'user-joined', userId: number } 
  | { type: 'user-left', userId: number };

const pubsub = new PubSub<Message>('chat-channel');

// Send a message when a button is clicked
document.getElementById('sendBtn')?.addEventListener('click', () => {
    const userId = 42;
    pubsub.broadcast({ type: 'chat-message', content: 'Hello from this tab!', userId });
});

// Listen for messages in other tabs
(async () => {
    for await (const message of pubsub) {
        switch (message.type) {
            case 'chat-message':
                console.log(`User ${message.userId} says: ${message.content}`);
                break;
            case 'user-joined':
                console.log(`User ${message.userId} joined the chat`);
                break;
            case 'user-left':
                console.log(`User ${message.userId} left the chat`);
                break;
        }
    }
})();

Now, whenever a user sends a message from one tab, all other tabs will receive and display it, and you can also notify when users join or leave the chat.

Why This is Useful

This PubSub system is especially helpful when building distributed systems or multi-window browser apps where you need a seamless way to share data without relying on external servers. Here’s why you might use it:

Cross-process: Share data between Node.js processes, Deno workers, or Bun worker threads.
Cross-tab in browsers: Synchronize state or events between browser tabs (e.g., messaging apps, shared state).
Type-safe messaging: The use of TypeScript’s type system allows for safe and predictable message handling, ensuring each message is handled correctly.

Conclusion

The PubSub class provides a simple, yet powerful solution for broadcasting events across multiple contexts, whether that's processes in Node.js, worker threads in Deno, or browser tabs. By utilizing the BroadcastChannel API and TypeScript’s discriminated union types, it can deliver cross-process or cross-tab event streaming with full type-safety, making it a versatile tool in both server-side and client-side environments.

With this in place, you can build more sophisticated, distributed systems or simply synchronize events across tabs or processes with minimal effort while enjoying all the benefits of TypeScript’s type system.

Asynchronously Iterating Over Event Emitters in TypeScript with Async Generators

Red John — Sun, 29 Sep 2024 21:59:10 +0000

Introduction

In modern web development, we often deal with events, whether it's handling incoming WebSocket messages, server-sent events (SSE), or data streams from services like Redis Pub/Sub. While Node.js provides event-driven capabilities, it lacks an out-of-the-box way to asynchronously iterate over events using for await...of loops.

In this post, I'll walk you through a simple yet powerful way to create an asynchronous event iterator using TypeScript and AsyncGenerator. This approach is designed to allow you to consume events from any kind of event emitter in a clean and predictable way, with full control over cancellation and cleanup logic.

The Use Case: Redis Pub/Sub

In one of my recent projects, I needed to listen to Redis Pub/Sub channels and dispatch server-sent events (SSE) asynchronously to connected clients. The challenge was handling incoming events without overwhelming the system while allowing the consumer to cancel the event stream at any time.

The solution? An event iterator that converts any event emitter (such as Redis Pub/Sub) into an asynchronous iterable. This allows us to process events in a controlled manner and gracefully handle cancellation when necessary.

Let’s dive into the implementation.

The Code

export type Context<T> = {
    emit: (value: T) => void;
    cancel: () => void;
};

export type CleanupFn = () => void | Promise<void>;

export type Subscriber<T> = (
    context: Context<T>,
) => void | CleanupFn | Promise<CleanupFn | void>;

export async function* createEventIterator<T>(
    subscriber: Subscriber<T>,
): AsyncGenerator<T> {
    const events: T[] = [];
    let cancelled = false;

    // Create a promise that resolves whenever a new event is added to the events array
    let resolveNext: (() => void) | null = null;

    const emit = (event: T) => {
        events.push(event);
        // If we are awaiting for a new event, resolve the promise
        if (resolveNext) {
            resolveNext();
            resolveNext = null;
        }
    };

    const cancel = () => {
        cancelled = true;
    };

    const unsubscribe = await subscriber({ emit, cancel });

    try {
        while (!cancelled) {
            // If there are events in the queue, yield the next event
            if (events.length > 0) {
                yield events.shift()!;
            } else {
                // Wait for the next event
                await new Promise<void>((resolve) => {
                    resolveNext = resolve;
                });
            }
        }

        // Process any remaining events that were emitted before cancellation.
        while (events.length > 0) {
            yield events.shift()!;
        }
    } finally {
        await unsubscribe?.();
    }
}

How It Works

This function accepts a subscriber function that you can hook into any event emitter or pub/sub system. The subscriber provides two essential methods:

emit: Allows the subscriber to push new events into the iterator.
cancel: Provides a way to signal that the iteration should stop.

The function returns an AsyncGenerator<T>, allowing you to iterate over events using a for await...of loop.

Breaking Down the Code

Context Object:
The Context<T> type provides an interface to emit new events or cancel the subscription. The subscriber uses this context to control the flow of events.
Event Queue:
The events: T[] array serves as a buffer to store emitted events. The generator will process these events one by one. If there are no events in the queue, it will wait for the next event to be emitted.
Emit Logic:
The emit function adds new events to the queue and resolves any pending promise (i.e., if the generator is waiting for new events).
Cancellation:
If the cancel function is called, it sets a flag (cancelled = true) to signal that the loop should exit. Any remaining events in the queue will still be processed before the generator completes.
Cleanup:
After cancellation, the generator will invoke the unsubscribe function (if provided) to perform any necessary cleanup. This is especially important for unsubscribing from external systems like Redis or cleaning up resources.

Example: Listening to Redis Pub/Sub

Let’s see how we can use this event iterator to listen to Redis Pub/Sub and asynchronously iterate over the incoming messages.

import Redis from 'ioredis';

function redisEventIterator(channel: string) {
    const client = new Redis();

    return createEventIterator<string>(({ emit, cancel }) => {
        const messageHandler = (channel: string, message: string) => {
            emit(message);
        };

        // Subscribe to the channel
        client.subscribe(channel);
        client.on('message', messageHandler);

        // Cleanup function to unsubscribe and disconnect
        return async () => {
            client.off('message', messageHandler);
            await client.unsubscribe(channel);
            await client.quit();
        };
    });
}

// Usage
(async () => {
    for await (const message of redisEventIterator('my-channel')) {
        console.log('New message:', message);

        // You can cancel the event stream if needed
        if (message === 'STOP') {
            break;
        }
    }
})();

In this example, we use createEventIterator to subscribe to a Redis Pub/Sub channel and asynchronously iterate over the messages. Each time a new message arrives, it is emitted into the generator, where we can process it in real-time. If a specific message (e.g., "STOP") is received, we break the loop and unsubscribe from Redis.

Example: Using EventEmitter

Here's how you can use createEventIterator with Node.js's EventEmitter:

import { EventEmitter } from 'events';

function eventEmitterIterator(emitter: EventEmitter, eventName: string) {
    return createEventIterator<string>(({ emit, cancel }) => {
        const eventHandler = (data: string) => emit(data);

        emitter.on(eventName, eventHandler);

        // Cleanup function to remove the listener
        return () => {
            emitter.off(eventName, eventHandler);
        };
    });
}

// Usage
(async () => {
    const emitter = new EventEmitter();

    // Simulate event emissions
    setTimeout(() => emitter.emit('data', 'First event'), 1000);
    setTimeout(() => emitter.emit('data', 'Second event'), 2000);
    setTimeout(() => emitter.emit('data', 'STOP'), 3000);

    for await (const event of eventEmitterIterator(emitter, 'data')) {
        console.log('Received event:', event);

        if (event === 'STOP') {
            break;
        }
    }
})();

In this example:

We use EventEmitter to emit events, which are captured by createEventIterator.
The iterator listens for the 'data' event and processes it asynchronously.
Similar to the Redis example, we can stop the iteration when a specific event ('STOP') is received.

Benefits of This Approach

Asynchronous Control: By leveraging the AsyncGenerator, we can handle events asynchronously, process them at our own pace, and pause processing when needed.
Cancellation: The ability to cancel the event stream at any time makes this approach flexible, especially in real-world scenarios where connections may need to be closed gracefully.
General-Purpose: This iterator can be used for any event emitter or Pub/Sub system, making it versatile for different applications.

Conclusion

Event-driven architectures are a cornerstone of many modern web applications, but they can become tricky to manage when we need to control the flow of events asynchronously. With the power of AsyncGenerator in TypeScript, you can build elegant solutions like this event iterator, making your event-handling code cleaner and easier to maintain.

I hope this post helps you get started with async iterators for your own event emitters. If you have any questions or thoughts, feel free to share them in the comments!

DEV Community: Red John

Building a Distributed PubSub Class with Cross-Process and Cross-Tab Capabilities

The PubSub Class Design

The Code

Key Components

Usage in Different Environments

1. Node.js / Bun Cluster with Typed Messages

2. Deno Workers with --parallel and Typed Events

3. Cross-Tab Communication in the Browser

Why This is Useful

Conclusion

Asynchronously Iterating Over Event Emitters in TypeScript with Async Generators

Introduction

The Use Case: Redis Pub/Sub

The Code

How It Works

Breaking Down the Code

Example: Listening to Redis Pub/Sub

Example: Using EventEmitter

Benefits of This Approach

Conclusion

The `PubSub` Class Design

2. Deno Workers with `--parallel` and Typed Events