mqemitter-valkey

A multi-process message emitter backed by Valkey Pub/Sub.

MQEmitter is an in-memory publish/subscribe emitter with MQTT-style topic matching (+ for one level, # for many). On its own it only reaches subscribers inside the same Node.js process. mqemitter-valkey lifts that limitation: every emit is relayed through Valkey Pub/Sub, so a message published by one process is delivered to matching subscribers in every process connected to the same Valkey. It opens one connection for publishing and one for subscribing (via iovalkey), encodes packets with MessagePack, and de-duplicates by packet id so a subscriber never sees the same message twice.

Install

npm install mqemitter-valkey

Usage

const valkey = require('mqemitter-valkey')

const mq = valkey({
  port: 6379,
  host: '127.0.0.1',
  password: 'my secret',
  db: 4
})

mq.on('hello/world', function (message, cb) {
  console.log(message)
  cb() // call the callback when you are done; never pass an error
})

// topic is mandatory
mq.emit({ topic: 'hello/world', payload: 'or any other fields' }, function () {
  // the emitter never returns an error
})

The same mq running in another process — pointed at the same Valkey — receives the message.

Topics and wildcards

Topics are /-separated by default. Two wildcards are supported when subscribing:

Wildcard	Matches	Example subscription	Matches topic
+	exactly one level	hello/+	hello/world
#	zero or more levels	hello/#	hello/a/b

Wildcard subscriptions are translated to Valkey pattern subscriptions (PSUBSCRIBE) under the hood; exact topics use plain SUBSCRIBE.

Options

The constructor accepts all iovalkey connection options (host, port, password, db, tls, …), plus:

Option	Description	Default
connectionString	A single connection URI instead of discrete options.	—
pubConn	Reuse an existing Valkey client for publishing.	new client
subConn	Reuse an existing Valkey client for subscribing.	new client
concurrency	Maximum number of messages processed concurrently (from MQEmitter).	0
maxLRUCache	Maximum number of packet ids kept for de-duplication.	10000
ttlLRUCache	Time-to-live of the de-duplication cache, in milliseconds.	60000

See MQEmitter for the full base API (on, emit, removeListener, close).

How it works

Each instance holds two Valkey connections. emit MessagePack-encodes the packet (with a unique id) and PUBLISHes it to a channel named after the topic. The subscribing connection receives every matching message, decodes it, checks the id against a short-lived LRU cache to drop duplicates, and re-emits it locally through MQEmitter's matcher — which then invokes the right listeners. Internal Sentinel chatter (__sentinel__:hello) is ignored.

TypeScript

Type declarations are included (mqemitter-valkey.d.ts) — no separate @types package needed.