@wrnexus/pubsub

Topic-based publish/subscribe with a pluggable driver — in-process by default, Redis for cross-process messaging.

Part of the WRNexusJS framework — an SSR-first, Bun-native full-stack web framework.

Overview

@wrnexus/pubsub is a small server-side pub/sub bus. You publish messages to a topic and subscribe with topic patterns; handlers fire for matching topics. The default driver keeps everything in-process, and you can swap in the Redis driver (@wrnexus/pubsub/redis) to fan messages out across processes or hosts. It also backs @wrnexus/core's realtime bridge for horizontal scaling.

bun add @wrnexus/pubsub

Private package — the machine must be authenticated to the wrnexus npm org

(a read token in ~/.npmrc). Requires Bun (Node is not supported).

API

createPubSub(driver?): PubSub

Creates a bus over a driver. Defaults to memoryDriver() (in-process).

interface PubSub {
  publish<T = unknown>(topic: string, message: T): Promise<void>;
  subscribe<T = unknown>(pattern: string, handler: Handler<T>): () => void;
  close(): Promise<void>;
}

type Handler<T = unknown> = (message: T, topic: string) => void | Promise<void>;

• publish(topic, message) — resolves once the driver and in-memory async handlers finish.
• subscribe(pattern, handler) — returns an unsubscribe function.
• close() — idempotently rejects new work, clears local subscriptions, and closes the driver.

Pattern matching

Subscription patterns match in three ways:

• Exact — "order:created" matches only that topic.
• Prefix — "order:*" matches any topic starting with "order:".
• Everything — "*" matches all topics.

memoryDriver(): PubSubDriver

The default in-process driver. Handlers are invoked synchronously (fire-and-forget for async handlers) whenever a published topic matches a registered pattern.

interface PubSubDriver {
  publish(topic: string, message: unknown): void | Promise<void>;
  subscribe(pattern: string, handler: Handler): () => void;
}

@wrnexus/pubsub/redis — redisDriver(url?)

A cross-process driver backed by Redis. It speaks RESP over a raw TCP socket via Bun.connect, so it adds no npm dependency. url defaults to $REDIS_URL, then redis://localhost:6379. The URL may carry a password and a database index (e.g. redis://:secret@host:6379/2).

function redisDriver(url?: string, options?: RedisDriverOptions): PubSubDriver & { close(): void };

• Exact topics use Redis SUBSCRIBE; wildcard patterns (ns:*, *) use

PSUBSCRIBE, whose glob semantics line up with this library's matching.

• Messages are JSON-stringified on publish and JSON.parsed on receipt; a payload

that isn't valid JSON is delivered as the raw string.

• close() tears down both the subscriber and publisher connections.
• Lost sockets reconnect with bounded exponential backoff and active subscriptions

are replayed. maxPending bounds unavailable-connection writes (default 1000); reconnectDelayMs and reconnectMaxDelayMs tune recovery (100ms/5000ms).

RESP codec (internal)

redis.ts uses a minimal RESP implementation exported from resp.ts (encodeCommand, parseReply, concat, and the RespValue type). These are implementation details of the Redis driver, not part of the public package entry.

Usage

In-process (default):

import { createPubSub } from "@wrnexus/pubsub";

const bus = createPubSub();

const off = bus.subscribe("order:*", (msg, topic) => {
  console.log(topic, msg);
});

await bus.publish("order:created", { id: 7 });

off(); // unsubscribe

Cross-process with Redis:

import { createPubSub } from "@wrnexus/pubsub";
import { redisDriver } from "@wrnexus/pubsub/redis";

const driver = redisDriver("redis://localhost:6379");
const bus = createPubSub(driver);

bus.subscribe("order:*", (msg, topic) => {
  // received on any app process subscribed to this pattern
});

await bus.publish("order:created", { id: 7 });

// on shutdown (also closes the driver)
await bus.close();

Requirements / Notes

• Bun-only. The Redis driver depends on Bun.connect; it throws

redisDriver requires the Bun runtime (Bun.connect). outside Bun. The default in-memory driver has no runtime dependencies.

• The Redis driver reads REDIS_URL from the environment when no url is passed.
• Backs [@wrnexus/core](../core)'s realtime bridge for horizontal scaling.
• No external npm dependencies — the Redis client is a self-contained RESP codec.