@snps/message-bus

Zero-dependency message bus system for decoupled, event-driven communication.

Installation

pnpm install @snps/message-bus

API

MessageBus

The MessageBus class provides a central communication hub for publishing and subscribing to messages.

publish<T>(type: string, payload: T, options?: PublishOptions)

Publishes a message to the bus.

import { MessageBus } from '@snps/message-bus';

const bus = new MessageBus();
bus.publish('user.created', { id: 1, name: 'John Doe' });

subscribe<T>(pattern: string, handler: MessageHandler<T>, options?: SubscribeOptions)

Subscribes to messages of a specific type or pattern.

import { MessageBus } from '@snps/message-bus';

const bus = new MessageBus();
const unsubscribe = bus.subscribe('user.created', (payload) => {
  console.log('User created:', payload);
});

request<T, R>(type: string, payload: T, timeout?: number)

Publishes a message and waits for a response.

import { MessageBus } from '@snps/message-bus';

const bus = new MessageBus();
const response = await bus.request('question', { question: 'What is the meaning of life?' });

respond<T>(message: Message, payload: T)

Responds to a request.

import { MessageBus } from '@snps/message-bus';

const bus = new MessageBus();
bus.subscribe('question', (payload, message) => {
  bus.respond(message, { answer: 42 });
});

use(middleware: Middleware)

Adds middleware for message processing.

import { MessageBus } from '@snps/message-bus';

const bus = new MessageBus();
bus.use((message, next) => {
  console.log('Middleware:', message.type);
  next();
});

getDeadLetterQueue(): Message[]

Gets the dead letter queue (failed messages).

const deadLetterQueue = bus.getDeadLetterQueue();

getHistory(): Message[]

Gets the message history.

const history = bus.getHistory();

replay(messageId: string)

Replays a message from history.

bus.replay('msg_123');

clear()

Clears all subscriptions and queues.

bus.clear();

getStats()

Gets statistics about the message bus.

const stats = bus.getStats();

getGlobalBus(): MessageBus

Provides a global singleton bus instance.

import { getGlobalBus } from '@snps/message-bus';

const globalBus = getGlobalBus();

resetGlobalBus()

Resets the global bus instance.

import { resetGlobalBus } from '@snps/message-bus';

resetGlobalBus();

createPublisher<T>(bus: MessageBus, type: string)

Utility to create a typed publisher.

import { MessageBus, createPublisher } from '@snps/message-bus';

const bus = new MessageBus();
const publishUserCreated = createPublisher<{ id: number, name: string }>(bus, 'user.created');
publishUserCreated({ id: 1, name: 'John Doe' });

createSubscriber<T>(bus: MessageBus, type: string, handler: MessageHandler<T>, options?: SubscribeOptions)

Utility to create a typed subscriber.

import { MessageBus, createSubscriber } from '@snps/message-bus';

const bus = new MessageBus();
const unsubscribe = createSubscriber<{ id: number, name: string }>(bus, 'user.created', (payload) => {
  console.log('User created:', payload);
});

createRequestHandler<T, R>(bus: MessageBus, type: string, handler: (payload: T, message: Message<T>) => R | Promise<R>)

Utility to create a request handler.

import { MessageBus, createRequestHandler } from '@snps/message-bus';

const bus = new MessageBus();
createRequestHandler<{ question: string }, { answer: number }>(bus, 'question', (payload) => {
  return { answer: 42 };
});