W WRNexusJS
Realtime

@wrnexus/pubsub

In-process and Redis-backed publish/subscribe.

bun add @wrnexus/pubsub@0.2.15
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.

Installation

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;
}

type Handler<T = unknown> = (message: T, topic: string) => void | Promise<void>;
  • publish(topic, message) — resolves once the driver has dispatched the message.
  • subscribe(pattern, handler) — returns an unsubscribe function.

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/redisredisDriver(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): 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.

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
driver.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.

Complete TypeScript API

This declaration is generated from the exact published package and lists its exported functions, classes, interfaces, and types.

/**
 * @wrnexus/pubsub — topic-based publish/subscribe with a pluggable driver.
 * The default is in-process; swap in a Redis/NATS driver for cross-instance
 * messaging (it also backs @wrnexus/core's realtime bridge).
 *
 *   const bus = createPubSub();
 *   const off = bus.subscribe("order:*", (msg, topic) => {...});
 *   await bus.publish("order:created", { id: 7 });
 *
 * Subscriptions match exact topics, "ns:*" prefixes, and "*" (everything).
 */
type Handler<T = unknown> = (message: T, topic: string) => void | Promise<void>;
interface PubSubDriver {
    publish(topic: string, message: unknown): void | Promise<void>;
    subscribe(pattern: string, handler: Handler): () => void;
}
interface PubSub {
    publish<T = unknown>(topic: string, message: T): Promise<void>;
    subscribe<T = unknown>(pattern: string, handler: Handler<T>): () => void;
}
/** In-process pub/sub driver (default). */
declare function memoryDriver(): PubSubDriver;
/** Create a pub/sub bus over a driver (in-memory by default). */
declare function createPubSub(driver?: PubSubDriver): PubSub;

export { type Handler, type PubSub, type PubSubDriver, createPubSub, memoryDriver };

Examples

Copy-ready examples taken from this package's published documentation.

Example 1

bun add @wrnexus/pubsub

Example 2

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

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

Example 3

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

Example 4

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