Pubsub

Cluster-aware pub/sub channel for in-process and cross-replica messaging. Three backends share one operator API:

local — in-process Map>; publish dispatches synchronously before returning. Single-node deploys pay zero coordination overhead. cluster — shared _blamejs_pubsub_messages table polled at pollIntervalMs; publish writes a row + dispatches locally; other nodes pick up rows via id > lastSeenId AND publishedBy <> selfNodeId. The default for any b.cluster-aware deploy. redis — Redis PUB/SUB on lib/redis-client.js. One connection enters subscribe mode (demultiplexed via setOnPushMessage); publish goes through a separate command-mode connection.

Pattern subscribe accepts glob-style topic matchers (* matches one .-delimited segment, ** matches any suffix). Channel names cap at 1 KiB to defeat pathological matcher inputs. Subscribe / unsubscribe is ref-counted across local handlers so the remote backend only carries one subscription per scoped channel.

Local dispatch always happens BEFORE publish() resolves regardless of backend — same-node subscribers see the payload with near-zero latency. Handler errors are caught and logged via the boot logger; they never abort dispatch to siblings. Bad-shape remote payloads drop silently after a logged warning so one malformed cross-node message can't poison the local handler chain.

b.pubsub.create(opts) #

{
  backend:         "local" | "cluster" | "redis" | object  // default "local"
  cluster:         object,                                 // required for backend "cluster"
  pollIntervalMs:  number,                                 // cluster poll cadence; default 100ms
  retentionMs:     number,                                 // cluster row retention; default 60_000
  pruneEveryMs:    number,                                 // cluster prune cadence; default 300_000
  redisUrl:        string,                                 // required for backend "redis"
  redisPassword:   string,
  redisUsername:   string,
  redisTls:        boolean,
  redisCa:         string | Buffer,
  redisServername: string,
  topicPrefix:     string,                                 // scopes every channel as `:`
  audit:           boolean,                                // default false; emit system.pubsub.publish
}

Build a pub/sub instance bound to one of the supported backends. Returned object exposes subscribe(channel, handler) / subscribePattern(pattern, handler) for receive, unsubscribe(token) for cleanup, publish(channel, payload) for fan-out, and close() for teardown. Tokens returned by subscribe are opaque records; pass them back to unsubscribe verbatim.

Throws PubsubError("UNKNOWN_BACKEND") when opts.backend is not one of "local", "cluster", "redis", or a custom backend object implementing { publishRemote, start, stop }. Throws PubsubError("BAD_BACKEND") when a custom backend object is missing those methods.

var ps = b.pubsub.create({ backend: "local" });

var token = ps.subscribe("user.created", function (payload, ev) {
  console.log(ev.channel, ev.source, payload.id);
  // → user.created local 42
});

await ps.publish("user.created", { id: 42 });
ps.unsubscribe(token);
await ps.close();

// Glob-style topic matchers: '*' matches one segment, '**' any suffix.
var ps = b.pubsub.create({ backend: "local" });

ps.subscribePattern("orders.*.created", function (payload, ev) {
  console.log(ev.channel);
  // → orders.eu.created
});

ps.subscribePattern("audit.**", function (payload, ev) {
  console.log(ev.channel);
  // → audit.security.login.failed
});

await ps.publish("orders.eu.created", { orderId: "ord_1" });
await ps.publish("audit.security.login.failed", { userId: "u_7" });
await ps.close();

Last updated 2026-08-08T16:39:49.652Z by seeder.