@wrnexus/pubsub
In-process and Redis-backed publish/subscribe.
bun add @wrnexus/pubsub@0.2.15Topic-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/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): PubSubDriver & { close(): void };
- Exact topics use Redis
SUBSCRIBE; wildcard patterns (ns:*,*) use - Messages are JSON-stringified on publish and
JSON.parsed on receipt; a payload close()tears down both the subscriber and publisher connections.
PSUBSCRIBE, whose glob semantics line up with this library's matching.
that isn't valid JSON is delivered as the raw string.
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 - The Redis driver reads
REDIS_URLfrom the environment when nourlis passed. - Backs [
@wrnexus/core](../core)'s realtime bridge for horizontal scaling. - No external npm dependencies — the Redis client is a self-contained RESP codec.
redisDriver requires the Bun runtime (Bun.connect). outside Bun. The default in-memory driver has no runtime dependencies.
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/pubsubExample 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 };