@marianmeres/pubsub

@marianmeres/pubsub

Lightweight, type-safe publish-subscribe with zero dependencies.

Features

• Typed events — optional generic event map for compile-time safety
• Wildcard subscriptions — subscribe to every event with "*"
• Async-aware — rejected promises from async subscribers are routed to your error handler
• Safe re-entrancy — subscribing/unsubscribing inside a callback never affects the in-flight publish
• Memory-efficient — empty topics are auto-cleaned
• using-friendly — the unsubscriber implements Symbol.dispose
• Destructuring-safe — methods are pre-bound, so const { publish } = createPubSub() just works

Install

# Deno
deno add jsr:@marianmeres/pubsub

# npm
npm install @marianmeres/pubsub

Basic Usage

import { createPubSub } from '@marianmeres/pubsub';

const { publish, subscribe, subscribeOnce, unsubscribe, unsubscribeAll } = createPubSub();

// Create a subscription (returns an unsubscribe function)
const unsub = subscribe('foo', console.log);

// Publish data
publish('foo', 'bar'); // logs 'bar'

// Unsubscribe — calling again is a safe no-op
unsub();

// Alternative ways to unsubscribe
unsubscribe('foo', console.log);
unsubscribeAll('foo');

// Now this is a no-op as no subscription exists
publish('foo', 'baz');

Typed Events

Pass an event map to get full type-safety on publish and subscribe:

import { createPubSub } from '@marianmeres/pubsub';

type Events = {
  'user:login':  { id: number; name: string };
  'user:logout': { id: number };
  'tick':        number;
};

const ps = createPubSub<Events>();

ps.publish('user:login', { id: 1, name: 'Ada' }); // ✓
// ps.publish('user:login', { id: 1 });           // ✗ TypeScript error
// ps.publish('unknown', 1);                      // ✗ TypeScript error

ps.subscribe('tick', (n) => {
  // n is inferred as number
  console.log(n + 1);
});

ps.subscribe('*', ({ event, data }) => {
  // event: keyof Events; data: union of all event payloads
});

Advanced Usage

Subscribe Once

Subscribe to an event that auto-unsubscribes after the first delivery. The unsubscribe happens before your callback runs, so re-entrant publishes from within the callback can never re-trigger it.

const ps = createPubSub();

ps.subscribeOnce('init', (data) => {
  console.log('Initialized:', data);
});

ps.publish('init', { ready: true }); // logs once
ps.publish('init', { ready: true }); // no-op

Subscribe to Multiple Topics

const unsub = ps.subscribeMany(['user:login', 'user:logout'], (data) => {
  console.log('user event', data);
});

unsub(); // removes all of them at once

Wildcard Subscriptions

Subscribe to all events using the "*" topic. Wildcard subscribers receive an envelope:

ps.subscribe('*', ({ event, data }) => {
  console.log(`event "${event}" published with`, data);
});

ps.publish('user:login', { userId: 123 });
// → event "user:login" published with { userId: 123 }

Note — "*" is reserved for subscribers. Calling publish('*', …) throws.

using Statement (ES2024 Explicit Resource Management)

The unsubscriber returned by subscribe, subscribeOnce, and subscribeMany implements Symbol.dispose. In environments that support using, subscriptions can scope themselves automatically:

{
  using sub = ps.subscribe('foo', (data) => console.log(data));
  ps.publish('foo', 'bar');
  // sub is auto-unsubscribed when the block exits
}

Async Subscribers

Subscribers may be async. Rejections are routed to your error handler the same way synchronous throws are — they will not surface as unhandled rejections.

const ps = createPubSub({
  onError: (err, topic, isWildcard) => log.error({ err, topic, isWildcard }),
});

ps.subscribe('save', async (data) => {
  await db.write(data);          // resolved value is ignored
  if (!data.ok) throw new Error('bad payload'); // → onError
});

Check Subscription Status

const cb = (data) => console.log(data);
ps.subscribe('foo', cb);

ps.isSubscribed('foo', cb);                  // true
ps.subscribe('*', cb);
ps.isSubscribed('bar', cb);                  // true (via wildcard)
ps.isSubscribed('bar', cb, false);           // false (excluding wildcard)

Introspection

ps.subscriberCount();        // total subscribers across all topics (incl. "*")
ps.subscriberCount('foo');   // direct subscribers on "foo"
ps.hasSubscribers('foo');    // true if "foo" has direct subscribers
ps.topics();                 // string[] — topics with at least one subscriber

Custom Error Handling

By default, errors thrown by subscribers (and rejections from async subscribers) are logged to console.error. Customize via the onError option:

// Silent mode
const ps = createPubSub({ onError: () => {} });

// Send to your logger
const ps = createPubSub({
  onError: (error, topic, isWildcard) => {
    myLogger.error('subscriber error', { error, topic, isWildcard });
  },
});

onError receives:

• error: Error — the thrown error or rejection reason
• topic: string — the topic being published
• isWildcard: boolean — true if the failing subscriber was attached to "*"

API Reference

For the full reference, see API.md.

Quick Reference

Method	Description
createPubSub<Events>(options?)	Factory returning a typed PubSub<Events>
new PubSub<Events>(options?)	Constructor
publish(topic, data?)	Publishes data; throws if topic === "*"
subscribe(topic, cb)	Subscribes; returns an Unsubscriber (Disposable)
subscribeOnce(topic, cb)	One-shot subscription
subscribeMany(topics, cb)	One callback for many topics
unsubscribe(topic, cb?)	Removes a callback or all subscribers of topic
unsubscribeAll(topic?)	Clears all subscribers of topic, or every topic
isSubscribed(topic, cb, considerWildcard?)	Subscription check
subscriberCount(topic?)	Count of subscribers (per topic, or total)
hasSubscribers(topic)	True if topic has direct subscribers
topics()	Topics with at least one subscriber

TypeScript

Full type support, including a generic event map:

import { PubSub, type PubSubOptions, type Subscriber, type Unsubscriber, type WildcardEnvelope }
  from '@marianmeres/pubsub';

type Events = { 'msg': string };

const ps = new PubSub<Events>({
  onError: (error, topic, isWildcard) =>
    console.warn(`[${isWildcard ? 'wildcard' : 'direct'}] ${topic}:`, error),
});

Important Notes

• Subscribers run synchronously in registration order. Async subscribers are not awaited; their rejections are routed to onError.
• The subscriber list is snapshotted before each publish — adding subscribers inside a callback never affects the in-flight publish; removing subscribers inside a callback does not suppress already-snapshotted callees for the in-flight publish.
• Subscriber errors are caught and routed to onError; other subscribers continue to run.
• Empty topics are auto-cleaned after the last subscriber leaves.
• The wildcard "*" is reserved for subscribers; publish('*', …) throws.
• Methods are bound in the constructor — destructuring createPubSub() is safe.

Migration from 2.x

Version 3 is a breaking release. See CHANGELOG (or the release notes) for the full list. Notable observable behavior changes:

• publish('*', …) now throws (was silent and inconsistent).
• subscribeOnce now fires exactly once, even under re-entrant publish.
• Subscribers added during a publish no longer fire in the same publish (snapshot semantics).
• Subscribers removed during a publish are still delivered for the in-flight publish (snapshot semantics).
• Async subscriber rejections now go through onError (previously: unhandled).
• unsubscribeAll() returns false when nothing was actually removed (previously always true).
• Unsubscriber return type narrowed from () => void | boolean to () => void.
• __dump() now returns a defensive copy of the internal sets.

Additive (non-breaking-style) additions: subscribeMany, subscriberCount, hasSubscribers, topics, generic PubSub<TEvents>, Symbol.dispose on the unsubscriber.

Package Identity

• Name: @marianmeres/pubsub
• Author: Marian Meres
• Repository: https://github.com/marianmeres/pubsub
• License: MIT