223 lines
18 KiB
Plaintext
223 lines
18 KiB
Plaintext
page wrnexuspubsub {
|
|
seo {
|
|
title = "@wrnexus/pubsub"
|
|
description = "In-process and Redis-backed publish/subscribe."
|
|
}
|
|
|
|
view {
|
|
<div class="docs-shell">
|
|
<a href="#main" class="skip-link">Skip to content</a>
|
|
<header class="topbar">
|
|
<a class="brand" href="/"><span>W</span> WRNexusJS</a>
|
|
<nav aria-label="Primary"><a href="/getting-started">Get started</a><a href="/packages">Packages</a><a href="https://component.wrnexusjs.dev/">Components</a><a href="/language">Language</a><a href="/architecture">Architecture</a></nav>
|
|
<div class="topbar-actions"><a class="preview-pill" href="/access">Private preview · v0.8.8</a><button data-wire-theme-toggle class="theme-button" aria-label="Toggle color theme" title="Toggle color theme">◐</button></div>
|
|
</header>
|
|
<div class="mobile-doc-nav"><details><summary>Browse documentation</summary><nav><a href="/getting-started">Get started</a><a href="/packages">Packages</a><a href="https://component.wrnexusjs.dev/">Components</a><a href="/language">Language</a><a href="/architecture">Architecture</a><a href="/tutorial">Tutorial</a><a href="/guides/project-structure">Guides</a><a href="/examples">Examples</a><a href="/search">Search</a></nav></details></div>
|
|
<main id="main" class="portal-main docs-layout docs-layout--navigation"><aside class="docs-section-menu" aria-label="Documentation sections"><nav>
|
|
<section><strong>Get started</strong><a href="/getting-started">Quick start</a><a href="/tutorial">Tutorial</a><a href="/guides/full-stack-auth-demo">Auth dashboard demo</a><a href="/guides/configuration-and-profiles">Configuration</a></section>
|
|
<section><strong>Build</strong><a href="/guides/pages-and-components">Pages and components</a><a href="/guides/routing">Routing</a><a href="/guides/server-data">Server data</a><a href="/guides/forms-and-validation">Forms and validation</a><a href="/guides/database">Database</a></section>
|
|
<section><strong>Auth and security</strong><a href="/guides/authentication">Authentication</a><a href="/guides/authorization">Authorization</a><a href="/guides/security">Application security</a><a href="/security">Security policy</a></section>
|
|
<section><strong>Reference</strong><a href="/packages">Packages overview</a><a href="/packages/cli">CLI</a><a href="/language">.wrn language</a><a href="https://component.wrnexusjs.dev/">Components</a><a href="/examples">Examples</a></section>
|
|
<section class="package-menu"><strong>Packages</strong><a href="/packages/ai">@wrnexus/ai</a><a href="/packages/auth">@wrnexus/auth</a><a href="/packages/authz">@wrnexus/authz</a><a href="/packages/benchmark">@wrnexus/benchmark</a><a href="/packages/cache">@wrnexus/cache</a><a href="/packages/captcha">@wrnexus/captcha</a><a href="/packages/cli">@wrnexus/cli</a><a href="/packages/compiler">@wrnexus/compiler</a><a href="/packages/content">@wrnexus/content</a><a href="/packages/core">@wrnexus/core</a><a href="/packages/csr">@wrnexus/csr</a><a href="/packages/db">@wrnexus/db</a><a href="/packages/dev-server">@wrnexus/dev-server</a><a href="/packages/dev-toolbar">@wrnexus/dev-toolbar</a><a href="/packages/encryption">@wrnexus/encryption</a><a href="/packages/graphql">@wrnexus/graphql</a><a href="/packages/helpers">@wrnexus/helpers</a><a href="/packages/i18n">@wrnexus/i18n</a><a href="/packages/identity">@wrnexus/identity</a><a href="/packages/image">@wrnexus/image</a><a href="/packages/jwt">@wrnexus/jwt</a><a href="/packages/language-server">@wrnexus/language-server</a><a href="/packages/mcp">@wrnexus/mcp</a><a href="/packages/mobile">@wrnexus/mobile</a><a href="/packages/native">@wrnexus/native</a><a href="/packages/oauth">@wrnexus/oauth</a><a href="/packages/observability">@wrnexus/observability</a><a href="/packages/playground">@wrnexus/playground</a><a href="/packages/plugin">@wrnexus/plugin</a><a href="/packages/pubsub">@wrnexus/pubsub</a><a href="/packages/pwa">@wrnexus/pwa</a><a href="/packages/queue">@wrnexus/queue</a><a href="/packages/reactive">@wrnexus/reactive</a><a href="/packages/realtime">@wrnexus/realtime</a><a href="/packages/router">@wrnexus/router</a><a href="/packages/security">@wrnexus/security</a><a href="/packages/ssr">@wrnexus/ssr</a><a href="/packages/store">@wrnexus/store</a><a href="/packages/styles">@wrnexus/styles</a><a href="/packages/syntax">@wrnexus/syntax</a><a href="/packages/test">@wrnexus/test</a><a href="/packages/tracking">@wrnexus/tracking</a><a href="/packages/typecheck">@wrnexus/typecheck</a><a href="/packages/ui">@wrnexus/ui</a><a href="/packages/uploader">@wrnexus/uploader</a><a href="/packages/validation">@wrnexus/validation</a></section>
|
|
</nav></aside><div class="docs-reading-column">
|
|
<article class="documentation prose standalone package-document"><nav class="breadcrumbs" aria-label="Breadcrumb"><a href="/">Home</a><span>/</span><a href="/packages">Packages</a><span>/</span><span aria-current="page">@wrnexus/pubsub</span></nav><section class="doc-intro"><span class="eyebrow">Realtime · Package reference</span><h1>@wrnexus/pubsub</h1><p>In-process and Redis-backed publish/subscribe.</p><div class="doc-meta"><span>v0.8.8</span><span>Private registry</span><span>Realtime</span></div><section id="access" class="access-callout"><h2>Install the package</h2><p>After WorkRoot approves private registry access, install the release-aligned package:</p><pre><code>bun add @wrnexus/pubsub@0.8.8</code><button type="button" class="copy-button" aria-label="Copy installation command">Copy</button></pre><p><a href="/access">Request preview access</a>. Never put registry tokens in source control.</p></section></section><section id="guide"><blockquote>Topic-based publish/subscribe with a pluggable driver — in-process by default, Redis for cross-process messaging.</blockquote>
|
|
<p>Part of the <strong>WRNexusJS</strong> framework — an SSR-first, Bun-native full-stack web framework.</p>
|
|
<h3 id="overview">Overview</h3>
|
|
<p><code>@wrnexus/pubsub</code> 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 (<code>@wrnexus/pubsub/redis</code>) to fan messages out across processes or hosts. It also backs <code>@wrnexus/core</code>'s realtime bridge for horizontal scaling.</p>
|
|
<pre data-language="bash"><code>bun add @wrnexus/pubsub</code></pre>
|
|
<blockquote>Private package — the machine must be authenticated to the <code>wrnexus</code> npm org</blockquote>
|
|
<blockquote>(a read token in <code>~/.npmrc</code>). Requires <strong>Bun</strong> (Node is not supported).</blockquote>
|
|
<h3 id="api">API</h3>
|
|
<h4 id="createpubsub-driver-pubsub"><code>createPubSub(driver?): PubSub</code></h4>
|
|
<p>Creates a bus over a driver. Defaults to <code>memoryDriver()</code> (in-process).</p>
|
|
<pre data-language="ts"><code>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>;</code></pre>
|
|
<ul>
|
|
<li><code>publish(topic, message)</code> — resolves once the driver and in-memory async handlers finish.</li>
|
|
<li><code>subscribe(pattern, handler)</code> — returns an unsubscribe function.</li>
|
|
<li><code>close()</code> — idempotently rejects new work, clears local subscriptions, and closes the driver.</li>
|
|
</ul>
|
|
<h4 id="pattern-matching">Pattern matching</h4>
|
|
<p>Subscription patterns match in three ways:</p>
|
|
<ul>
|
|
<li><strong>Exact</strong> — <code>"order:created"</code> matches only that topic.</li>
|
|
<li><strong>Prefix</strong> — <code>"order:*"</code> matches any topic starting with <code>"order:"</code>.</li>
|
|
<li><strong>Everything</strong> — <code>"*"</code> matches all topics.</li>
|
|
</ul>
|
|
<h4 id="memorydriver-pubsubdriver"><code>memoryDriver(): PubSubDriver</code></h4>
|
|
<p>The default in-process driver. Handlers are invoked synchronously (fire-and-forget for async handlers) whenever a published topic matches a registered pattern.</p>
|
|
<pre data-language="ts"><code>interface PubSubDriver {
|
|
publish(topic: string, message: unknown): void | Promise<void>;
|
|
subscribe(pattern: string, handler: Handler): () => void;
|
|
}</code></pre>
|
|
<h4 id="wrnexus-pubsub-redis-redisdriver-url"><code>@wrnexus/pubsub/redis</code> — <code>redisDriver(url?)</code></h4>
|
|
<p>A cross-process driver backed by Redis. It speaks RESP over a raw TCP socket via <code>Bun.connect</code>, so it adds <strong>no npm dependency</strong>. <code>url</code> defaults to <code>$REDIS_URL</code>, then <code>redis://localhost:6379</code>. The URL may carry a password and a database index (e.g. <code>redis://:secret@host:6379/2</code>).</p>
|
|
<pre data-language="ts"><code>function redisDriver(url?: string, options?: RedisDriverOptions): PubSubDriver & { close(): void };</code></pre>
|
|
<ul>
|
|
<li>Exact topics use Redis <code>SUBSCRIBE</code>; wildcard patterns (<code>ns:*</code>, <code>*</code>) use</li>
|
|
<p><code>PSUBSCRIBE</code>, whose glob semantics line up with this library's matching.</p>
|
|
<li>Messages are JSON-stringified on publish and <code>JSON.parse</code>d on receipt; a payload</li>
|
|
<p>that isn't valid JSON is delivered as the raw string.</p>
|
|
<li><code>close()</code> tears down both the subscriber and publisher connections.</li>
|
|
<li>Lost sockets reconnect with bounded exponential backoff and active subscriptions</li>
|
|
<p>are replayed. <code>maxPending</code> bounds unavailable-connection writes (default 1000); <code>reconnectDelayMs</code> and <code>reconnectMaxDelayMs</code> tune recovery (100ms/5000ms).</p>
|
|
</ul>
|
|
<h4 id="resp-codec-internal">RESP codec (internal)</h4>
|
|
<p><code>redis.ts</code> uses a minimal RESP implementation exported from <code>resp.ts</code> (<code>encodeCommand</code>, <code>parseReply</code>, <code>concat</code>, and the <code>RespValue</code> type). These are implementation details of the Redis driver, not part of the public package entry.</p>
|
|
<h3 id="usage">Usage</h3>
|
|
<p>In-process (default):</p>
|
|
<pre data-language="ts"><code>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</code></pre>
|
|
<p>Cross-process with Redis:</p>
|
|
<pre data-language="ts"><code>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();</code></pre>
|
|
<h3 id="requirements-notes">Requirements / Notes</h3>
|
|
<ul>
|
|
<li><strong>Bun-only.</strong> The Redis driver depends on <code>Bun.connect</code>; it throws</li>
|
|
<p><code>redisDriver requires the Bun runtime (Bun.connect).</code> outside Bun. The default in-memory driver has no runtime dependencies.</p>
|
|
<li>The Redis driver reads <code>REDIS_URL</code> from the environment when no <code>url</code> is passed.</li>
|
|
<li>Backs [<code>@wrnexus/core</code>](../core)'s realtime bridge for horizontal scaling.</li>
|
|
<li>No external npm dependencies — the Redis client is a self-contained RESP codec.</li>
|
|
</ul></section><section id="api" class="api"><h2>Complete TypeScript API</h2><p>Generated from the exact installed package declarations.</p><pre data-language="typescript"><code>import { Context } from '@wrnexus/core';
|
|
import { SubjectContext } from '@wrnexus/rpc';
|
|
|
|
interface MessageEnvelope<T = unknown> {
|
|
id: string;
|
|
topic: string;
|
|
data: T;
|
|
timestamp: number;
|
|
attempts: number;
|
|
}
|
|
interface ResilientPubSubOptions {
|
|
retries?: number;
|
|
retryDelayMs?: number;
|
|
onError?: (error: unknown, envelope: MessageEnvelope) => void;
|
|
}
|
|
declare function createResilientPubSub(driver: PubSubDriver, options?: ResilientPubSubOptions): PubSub;
|
|
interface PresenceMember {
|
|
id: string;
|
|
metadata?: Record<string, unknown>;
|
|
joinedAt: number;
|
|
expiresAt: number;
|
|
}
|
|
declare class PresenceChannel {
|
|
#private;
|
|
private readonly ttlMs;
|
|
private readonly now;
|
|
constructor(ttlMs?: number, now?: () => number);
|
|
touch(id: string, metadata?: Record<string, unknown>): PresenceMember;
|
|
leave(id: string): boolean;
|
|
list(): PresenceMember[];
|
|
prune(): number;
|
|
}
|
|
|
|
interface SubjectPubSub {
|
|
publish<T>(ctx: Context, topic: string, message: T): Promise<void>;
|
|
subscribe<T>(pattern: string, handler: (message: T, topic: string, subject?: SubjectContext) => void | Promise<void>): () => void;
|
|
}
|
|
/**
|
|
* Authenticated pub/sub envelope. The token uses a fixed, purpose-specific
|
|
* audience; subscribers verify it before exposing the message to a handler.
|
|
*/
|
|
declare function subjectPubSub(bus: PubSub): SubjectPubSub;
|
|
|
|
/**
|
|
* @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;
|
|
close?(): void | Promise<void>;
|
|
}
|
|
interface PubSub {
|
|
publish<T = unknown>(topic: string, message: T): Promise<void>;
|
|
subscribe<T = unknown>(pattern: string, handler: Handler<T>): () => void;
|
|
/** Stop new work, remove subscriptions, and close the backing driver. */
|
|
close(): Promise<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;
|
|
|
|
interface NatsClient {
|
|
publish(subject: string, data: Uint8Array): void | Promise<void>;
|
|
subscribe(subject: string, handler: (data: Uint8Array, subject: string) => void): () => void;
|
|
close?(): void | Promise<void>;
|
|
}
|
|
declare function natsDriver(client: NatsClient): PubSubDriver;
|
|
interface KafkaClient {
|
|
publish(topic: string, value: string): void | Promise<void>;
|
|
subscribe(pattern: string, handler: (value: string, topic: string) => void): () => void;
|
|
close?(): void | Promise<void>;
|
|
}
|
|
/** Kafka adapter contract; consumer-group/rebalance policy remains owned by the selected client. */
|
|
declare function kafkaDriver(client: KafkaClient): PubSubDriver;
|
|
|
|
export { type Handler, type KafkaClient, type MessageEnvelope, type NatsClient, PresenceChannel, type PresenceMember, type PubSub, type PubSubDriver, type ResilientPubSubOptions, type SubjectPubSub, createPubSub, createResilientPubSub, kafkaDriver, memoryDriver, natsDriver, subjectPubSub };
|
|
</code></pre></section><section id="examples" class="examples"><h2>Examples</h2><p>Copy-ready examples from the installed package documentation.</p><div class="example-grid"><article class="example-card"><h3>In-process (default)</h3><pre data-language="ts"><code>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</code></pre></article><article class="example-card"><h3>Cross-process with Redis</h3><pre data-language="ts"><code>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();</code></pre></article></div></section></article></div>
|
|
<aside class="on-this-page"><h2>On this page</h2><nav><a class="toc-level-2" href="#guide">Guide</a><a class="toc-level-3" href="#overview">Overview</a><a class="toc-level-3" href="#api">API</a><a class="toc-level-4" href="#createpubsub-driver-pubsub">createPubSub(driver?): PubSub</a><a class="toc-level-4" href="#pattern-matching">Pattern matching</a><a class="toc-level-4" href="#memorydriver-pubsubdriver">memoryDriver(): PubSubDriver</a><a class="toc-level-4" href="#wrnexus-pubsub-redis-redisdriver-url">@wrnexus/pubsub/redis — redisDriver(url?)</a><a class="toc-level-4" href="#resp-codec-internal">RESP codec (internal)</a><a class="toc-level-3" href="#usage">Usage</a><a class="toc-level-3" href="#requirements-notes">Requirements / Notes</a><a class="toc-level-2" href="#api">Complete API</a><a class="toc-level-2" href="#examples">Examples</a></nav></aside>
|
|
</main>
|
|
<footer><div class="footer-brand"><span class="footer-mark" aria-hidden="true">W</span><p><strong>WRNexusJS 0.8.8</strong><span>Complete API documentation generated from installed package declarations.</span></p></div><nav aria-label="Footer"><a href="/packages">All packages</a><a href="/getting-started">Get started</a><a href="/security">Security</a><a href="/support">Support</a><a href="/llms.txt">AI guide</a></nav><p class="footer-meta">Private Developer Preview · Bun-native</p></footer>
|
|
</div>
|
|
}
|
|
}
|