Snub

Redis-backed pub/sub messaging for Node.js services.

Snub lets your services talk to each other over Redis with minimal boilerplate. Send to one worker, broadcast to all, await replies, schedule delayed jobs — all with a clean, chainable API.

Features

• Mono — single delivery, one worker claims the message (load balanced)
• Poly — broadcast to every listener across all instances
• Replies — request/response patterns with awaitReply() or callback-based replyAt()
• Glob patterns — subscribe to user:* and match any channel under it
• Namespaced listeners — multiple handlers on the same event, removable by name
• Delayed jobs — schedule mono messages with sendDelay(), cancel with cancelDelay()
• Interceptors — middleware functions that run before each handler, can block or mutate
• Separate Redis connections — pub/sub and storage connections are kept independent

Installation

npm install snub

Setup

const Snub = require('snub');

const snub = new Snub({
  // Redis connection for pub/sub
  redisAuth: {
    host: 'localhost',
    port: 6379,
    password: '',
    db: 0,
  },

  // Optional: separate Redis connection for storage (defaults to redisAuth)
  redisStore: {
    host: 'localhost',
    port: 6379,
    db: 1,
  },

  timeout: 5000,          // reply timeout in ms (default: 5000)
  monoWait: 50,           // max jitter in ms before workers race to claim mono messages
  prefix: 'snub',         // Redis key prefix
  processDelayQueue: true, // set false on instances that should not poll the delay queue

  // Interceptor — runs before every handler. Return false to block.
  interceptor: async (payload, reply, listener, channel) => {
    if (listener === 'blocked-event') return false;
    return true;
  },

  // Stats — called after each user event is handled
  stats: ({ pattern, replyTime }) => {
    console.log(`${pattern} replied in ${replyTime}ms`);
  },
});

Configuration Reference

| Option | Type | Default | Description | |---|---|---|---| | redisAuth | object | string | null | ioredis options or URL for pub/sub connections | | redisStore | object | string | null | ioredis options or URL for storage. Falls back to redisAuth | | prefix | string | 'snub' | Redis key namespace prefix | | timeout | number | 5000 | Default reply timeout in ms | | monoWait | number | 50 | Max random wait (ms) before workers race to claim a mono message | | nsSeparator | string | '.' | Separator between event name and namespace | | delayResolution | number | 1000 | How often (ms) to poll the delay queue | | processDelayQueue | boolean | true | Set false to skip delay queue polling on this instance | | interceptor | function | function[] | passthrough | Run before each handler. Return false to block | | stats | function | no-op | Called after each user event is handled | | debug | boolean | false | Log internal events to console |

API

snub.on(pattern, handler)

Subscribe to an event or glob pattern. The handler signature is:

handler(payload, reply, channel, eventPattern)

• payload — the data sent with the event
• reply(data) — call to send a reply back to the emitter
• channel — the exact channel name the message was published on
• eventPattern — the pattern this listener was registered with

snub.once(pattern, handler)

Same as on, but automatically unsubscribes after the first delivery.

snub.off(pattern)

Unsubscribe all listeners for a pattern, or just a namespaced one:

snub.off('user:login');        // removes all listeners for 'user:login'
snub.off('user:login.myApp');  // removes only the 'myApp' namespace listener

snub.mono(channel, data) → EmitObject

Emits a single-delivery message. Only one listener across all instances receives it — whichever claims it first (with optional jitter via monoWait).

snub.poly(channel, data) → EmitObject

Broadcasts to every listener on every instance.

EmitObject methods

| Method | Description | |---|---| | .send(cb?) | Publish the message. cb(count) receives subscriber count. Returns Promise<count> | | .awaitReply(timeout?, expectedReplies?) | Await reply(ies). Mono resolves with a single value; poly resolves with an array | | .replyAt(callback, timeout?) | Callback-based reply handler. Chainable before .send() | | .sendDelay(seconds) | Schedule delivery after N seconds. Returns a job ID. Mono only |

snub.cancelDelay(jobId) → Promise<boolean>

Cancel a previously scheduled delay job. Returns true if found and cancelled, false if already fired or not found.

snub.close() → Promise<void>

Gracefully shuts down: clears the delay interval and quits all Redis connections.

Examples

Mono — single delivery with reply

Only one instance handles the message, even if many are listening:

snub.on('order:process', async (payload, reply) => {
  const result = await processOrder(payload);
  reply({ orderId: result.id, status: 'ok' });
});

const response = await snub.mono('order:process', { item: 'widget', qty: 3 }).awaitReply();
console.log(response); // { orderId: '...', status: 'ok' }

Poly — broadcast to all listeners

Every instance that has registered a handler receives the message:

snub.on('config:reload', (payload) => {
  reloadConfig(payload.version);
});

// Notify all instances to reload
snub.poly('config:reload', { version: '2.1.0' }).send();

Poly with replies

Collect responses from every instance:

snub.on('health:check', (payload, reply) => {
  reply({ host: os.hostname(), uptime: process.uptime() });
});

const results = await snub.poly('health:check').awaitReply();
console.log(results); // [{ host: 'web-1', uptime: 120 }, { host: 'web-2', uptime: 95 }]

Glob patterns

snub.on('game:*', (payload, reply, channel) => {
  console.log(`Received on channel: ${channel}`);
});

snub.poly('game:start', { gameId: 'abc' }).send();
snub.poly('game:end',   { gameId: 'abc' }).send();

Namespaced listeners

Register multiple handlers for the same event and remove them independently:

snub.on('user:login.analytics', (payload) => trackLogin(payload));
snub.on('user:login.audit',     (payload) => auditLog(payload));

// Remove only the analytics handler
snub.off('user:login.analytics');

// Remove all user:login handlers
snub.off('user:login');

Callback-based replies with replyAt

Useful when you want to handle each reply as it arrives rather than waiting for all of them:

snub.poly('worker:status', null)
  .replyAt((status) => {
    console.log('Worker replied:', status);
  }, 2000)
  .send();

Delayed delivery with sendDelay

Schedules a mono message for future delivery. The message is stored in Redis and dispatched by whichever Snub instance polls the queue first.

Note: sendDelay is fire-and-schedule. Replies are not supported — the original call context is gone by the time the message fires.

snub.on('reminder:send', (payload) => {
  sendReminderEmail(payload.userId, payload.message);
});

// Schedule delivery in 60 seconds
const jobId = await snub.mono('reminder:send', {
  userId: 'u_123',
  message: 'Your session is about to expire.',
}).sendDelay(60);

// Cancel it if no longer needed
const cancelled = await snub.cancelDelay(jobId);

To opt an instance out of polling the delay queue (useful in multi-process setups where only dedicated workers should dispatch delayed jobs):

new Snub({ processDelayQueue: false, ...redisConfig });

Interceptors

Interceptors run before every handler and can block, mutate, or log events:

const snub = new Snub({
  interceptor: [
    // Auth check
    async (payload, reply, listener) => {
      if (listener.startsWith('admin:') && !payload.token) {
        reply({ error: 'Unauthorized' });
        return false; // block
      }
      return true;
    },
    // Event logger
    async (payload, reply, listener) => {
      console.log(`[snub] ${listener}`, payload);
      return true;
    },
  ],
  ...redisConfig,
});

Using use for plugins

snub.use((snubInstance) => {
  snubInstance.on('*', (payload, reply, channel) => {
    metrics.increment('snub.events', { channel });
  });
});

Direct Redis Access

The underlying ioredis connections are exposed if you need direct Redis access:

snub.redis // storage connection (redisStore)
snub.pub   // publish connection
snub.sub   // subscribe connection

Requirements

• Node.js 14+
• A running Redis instance (v4+)

Ecosystem

• snub-http — HTTP API gateway over Snub
• snub-ws — WebSocket server over Snub
• snub-cron — Cron jobs distributed via Snub
• snub-store — Shared key-value store via Snub

Contributing

Issues and pull requests are welcome.

License

MIT