tyo-mq

TYO-MQ is a distributed messaging (pub/sub) service with socket.io.

TYO-MQ supports fire-and-forget pub/sub by default, plus opt-in durable delivery with ACK, retry, and dead-letter handling for messages that must survive disconnects or failed consumers.

See it in action: the tyo-mq-samples repo has four runnable mini-apps — a job queue with retry and DLQ, a browser chat, IoT telemetry over wildcard topics, and event-driven microservices — plus a cookbook with one self-contained script per feature. Clone, npm install, run.

Installation

npm install tyo-mq

Creating a messaging server

var MessageServer = require("tyo-mq").Server;

var mq = new MessageServer();
mq.start();

Creating a message producer

var Factory = require('tyo-mq').Factory,
    producer;

var mq = new Factory();  

mq.createProducer('testevent')
.then(function (p) {
    producer = p;

    // produce a default event with data {data: 'test'}
    producer.produce('test text from default event');

    // produce a different kind of event
    producer.produce('event2', {data: 'test text from event2'})
});

Creating a message subscriber

var Factory = require('tyo-mq').Factory,
    consumer;

var mq = new Factory();    

mq.createConsumer()
.then(function (c) {
    consumer = c;
    consumer.on('connect', function ()  {
        console.log('consumer\'s own connect listenr');
    });

    // subscribe 'event2'
    consumer.subscribe('event2', (data) => {
        console.log(data);
    });

    // subscribe 'testevent'
    consumer.subscribe('testevent', (data) => {
        console.log(data);
    });
});

Durable Delivery and ACK

Durable subscriptions store matching messages while the consumer is offline. ACK is not required by default: durable: true alone keeps the Phase 2 behavior, where replayed durable messages are removed immediately after delivery. When a subscription explicitly asks for ACK with ack, require_ack, or manual_ack, the server includes a msgId, waits for ACK {msgId}, retries on timeout, and moves exhausted messages to the realm DLQ.

consumer.subscribe(producer.name, 'task', function (data) {
    // auto-ACK after the handler resolves
}, {
    durable: true,
    ack: true,
    retry: { max_attempts: 3, delay: '5s', backoff: 'exponential' }
});

consumer.subscribe(producer.name, 'task', function (data, from, ack, raw) {
    doWork(data).then(function () {
        ack();
    });
}, {
    durable: true,
    manual_ack: true,
    ack_timeout: '30s',
    retry: { max_attempts: 5, delay: '2s' }
});

The storage backends expose deadLetter(msgId, reason), listDlq(realm), and discardDlq(msgId) for management tooling. The in-memory backend keeps DLQ entries until process exit; SQLite and Redis keep them in their configured stores.

Topics, Consumer Groups, and Broadcast

All routing features are opt-in; the default producer/event routing is unchanged.

Topic subscriptions ({ mode: 'topic' }) use MQTT-style hierarchical patterns: + matches exactly one level, # matches any number of trailing levels. Topic subscriptions match the produced event name from any producer. They compose with durable, ack, and retry like any other subscription.

consumer.subscribe('org/acme/machine/+/cmd', function (data, from, ack, raw) {
    console.log('command for', raw.event, ':', data);  // raw.event = concrete topic
}, { mode: 'topic' });

consumer.subscribe('org/acme/#', handler, { mode: 'topic', durable: true });

producer.produce('org/acme/machine/m-01/cmd', 'restart');

Consumer groups ({ group: 'name' }) share the load of an event or topic: each message is delivered to exactly one member of the group (round-robin), while ungrouped subscribers still receive every message.

workerA.subscribe('org/acme/machine/+/cmd', handler, { mode: 'topic', group: 'workers' });
workerB.subscribe('org/acme/machine/+/cmd', handler, { mode: 'topic', group: 'workers' });

Broadcast sends one message to every connected member of the producer's realm — including realm members with no matching subscription — or to every member of a group (one copy each, no load balancing):

producer.produce('announcement', data, { broadcast: 'realm' });
producer.produce('rollout', data, { broadcast: 'group', group: 'workers' });

Broadcast is fire-and-forget: it does not enqueue for offline durable subscribers.

Clustering

Multiple tyo-mq nodes can share one Redis: managed settings (realms, pre-shared keys, approved tokens) stay in sync across nodes, signed manager commands cannot be replayed against a peer node, durable subscribers can reconnect to any node, live messages are relayed so a producer on one node reaches subscribers on every node, and authorization requests can be decided from any node. Enable it in the settings file on every node:

{
  "storage": "redis",
  "storage_options": { "url": "redis://10.0.0.5:6379/0" },
  "cluster": { "enabled": true }
}

cluster.redis_url defaults to storage_options.url. Consumer groups are load-balanced per node, so a group's members should connect to the same node. See docs/CLUSTERING.md for the full setup guide and delivery semantics.

Observability (opt-in HTTP API)

A read-only HTTP surface can be enabled at startup, served on the same port as the socket server (no second port to firewall). It is off by default — without the option, no HTTP endpoint exists.

{ "http_api": { "enabled": true } }

| Endpoint | Auth | Description | |----------|------|-------------| | GET /health | none | Liveness for load balancers: status, version, uptime, cluster node id | | GET /api/metrics | Bearer admin token* | Prometheus text format: produced/delivered/queued/DLQ/ack-timeout counters, connection counts | | GET /api/stats | Bearer admin token | Per-realm producer/consumer totals and online counts, subscription counts | | GET /api/realms/{realm}/dlq | Bearer admin token | Dead-letter queue contents for the realm |

* When auth is disabled, no token is needed. http_api.metrics_auth: false opens /api/metrics for scrapers that cannot send a token (Prometheus itself supports authorization: { credentials: ... } in the scrape config, so the default is to require it). Individual endpoints can be switched off with health: false, metrics: false, or stats: false.

Management (write) operations intentionally stay on the signed socket command channel — the HTTP surface is read-only by design.

Demo

Start the TYO-MQ server

# Needs to set up the library (module) path
export NODE_PATH=`npm config get prefix`/lib/node_modules/
node -e 'require("tyo-mq/server")'

Customize Server Configuration

You can customize the server configuration including CORS settings by creating your own server file:

var Server = require('tyo-mq').Server;

var server = new Server({
    serveClient: false,
    pingInterval: 5000,
    pingTimeout: 10000,
    allowEIO3: true,
    // CORS configuration
    cors: {
        origin: "*",  // Allow all origins, or specify ["http://localhost:3000", "https://yourdomain.com"]
        methods: ["GET", "POST"],
        credentials: true
    },
    // WebSocket compression settings
    perMessageDeflate: {
        threshold: 2048,
        zlibDeflateOptions: {
            chunkSize: 8 * 1024,
        },
        zlibInflateOptions: {
            windowBits: 14,
            memLevel: 7,
        },
        clientNoContextTakeover: true,
        serverNoContextTakeover: true,
        serverMaxWindowBits: 10,
        concurrencyLimit: 20,
    },
});

server.start(8080); // Specify custom port

Authentication and Realms

Authentication is disabled by default for backwards compatibility. When enabled, clients must send AUTHENTICATION before registering as a producer or consumer. The built-in clients can do this automatically with a configured token:

var Server = require('tyo-mq').Server;
var Factory = require('tyo-mq').Factory;

var server = new Server({
    auth: {
        enabled: true,
        tokens: [
            { token: 'secret-acme-prod', realm: 'acme', role: 'producer' },
            { token: 'secret-acme-cons', realm: 'acme', role: 'consumer' }
        ]
    }
});
server.start();

var producerMq = new Factory({ auth: { token: 'secret-acme-prod' } });
var consumerMq = new Factory({ auth: { token: 'secret-acme-cons' } });

Supported roles are producer, consumer, both, manager, and admin. Producers, consumers, and subscriptions are scoped to the authenticated realm, so the same producer or consumer names can exist independently in different realms. manager is a per-realm administration role; admin spans all realms (realm: "*"). In addition to configured opaque tokens, the server can validate HS256 JWTs with auth.jwt_secret or delegate validation to an HTTP endpoint with auth.auth_url.

Connection authorization

Only manager connections always require manual authorization (an approved token, granted via the authorization request flow or configured directly). Other roles are governed by per-realm settings:

{
  "auth": {
    "enabled": true,
    "realms": {
      "acme": {
        "key": "consumer-pre-shared-key",
        "require_acceptance": true
      }
    }
  }
}

• consumer (and both) present the realm's pre-shared key. Connections without a realm, or to a realm without a key (or with require_key: false), are allowed automatically.
• producer (and both) must be accepted into the realm through the authorization request flow, unless the realm sets require_acceptance: false.

// consumer joins with the realm pre-shared key — no token needed
var consumerMq = new Factory({
    auth: { realm: 'acme', role: 'consumer', key: 'consumer-pre-shared-key' }
});

The pre-shared key and acceptance requirement are managed with the signed management commands set_realm_key and set_realm_acceptance — available from the interactive manager (npm run manager), the web manager UI (npm run manager:web), or Authorization.authManagementCommand().

When auth is enabled and no realm: "*", role: "admin" token is configured, the server creates one automatically and appends it to .env as TYO_MQ_ADMIN_TOKEN. The default server.js loads .env, so you can enable auth and let the first server start create the admin token:

printf 'TYO_MQ_AUTH_ENABLED=true\n' > .env
npm start

In another shell, verify the generated token can authenticate:

npm run auth:admin

The helper reads TYO_MQ_ADMIN_TOKEN from .env and sends AUTHENTICATION to the running server.

Managers do not need to send shared secrets to approve new client tokens. A client can submit a pending authorization request:

npm run auth:request -- \
  --realm tyolab \
  --role consumer \
  --client-id tyolab-agent-01 \
  --client-name "Tyolab Agent 01"

The command prints a generated client_token and request_id. A server admin can retrieve pending requests and approve or reject them with a signed proof derived from TYO_MQ_ADMIN_TOKEN:

npm run auth:manager -- next
npm run auth:manager -- approve <request_id> --role consumer
npm run auth:manager -- reject <request_id> --reason "unknown client"

An org or realm operator can use a scoped realm manager key instead:

{
  "auth": {
    "realms": {
      "tyolab": {
        "required": true,
        "manager_key": "realm-manager-shared-secret"
      }
    }
  }
}

TYO_MQ_REALM_MANAGER_KEY="realm-manager-shared-secret" \
  npm run auth:manager -- next --realm tyolab

TYO_MQ_REALM_MANAGER_KEY="realm-manager-shared-secret" \
  npm run auth:manager -- approve <request_id> --role consumer

The manager proof is an HMAC-SHA256 signature over the action, body, timestamp, and nonce. The admin token or realm manager key stays local to the manager script. A realm manager key can only poll and decide authorization requests in its configured realm; server-wide management commands still require the global admin token. Approved client tokens are added to the server's auth token list. When TYO_MQ_SETTINGS_FILE is configured, approved tokens are persisted to that file and survive restarts; otherwise they are runtime-only.

Approved client tokens can be revoked through the interactive manager or the signed management command revoke_token. Revocation can identify a token by token_hash, or by realm plus client_id.

The same flow is available as library calls:

var Authorization = require('tyo-mq').Authorization;

await Authorization.submitAuthorizationRequest({
    realm: 'tyolab',
    role: 'consumer',
    client_id: 'tyolab-agent-01',
    client_name: 'Tyolab Agent 01',
    client_token: 'client-secret',
    challenge_response: { ticket: 'INC-123' }
});

var next = await Authorization.nextAuthorizationRequest(process.env.TYO_MQ_ADMIN_TOKEN);
await Authorization.decideAuthorizationRequest(process.env.TYO_MQ_ADMIN_TOKEN, {
    request_id: next.request.request_id,
    approved: true,
    role: 'consumer'
});

var realmNext = await Authorization.nextRealmAuthorizationRequest(
    process.env.TYO_MQ_REALM_MANAGER_KEY,
    'tyolab'
);
await Authorization.decideRealmAuthorizationRequest(process.env.TYO_MQ_REALM_MANAGER_KEY, {
    request_id: realmNext.request.request_id,
    approved: true,
    role: 'consumer'
});

For interactive auth and realm management:

npm run manager

With Docker Compose, provide the admin token from the host and persist server settings through the bundled volume:

export TYO_MQ_ADMIN_TOKEN="$(openssl rand -hex 32)"
docker compose up -d tyo-mq
docker compose run --rm manager

CORS Options:

• origin: "*" - Allow all origins (development/testing)
• origin: ["http://localhost:3000"] - Allow specific origins (production)
• methods - Allowed HTTP methods
• credentials: true - Allow credentials in requests

Test Script

export NODE_PATH=`npm config get prefix`/lib/node_modules/
node -e 'require("tyo-mq/test")'

Browserify

This package supports being browserified. In order to browserify, please install two more extra packages:

npm install utf-8-validate bufferutil

Afterward,

browserify web/web.js -o web/client/tyo-mq-client.js

TODO list

• implement the message queuing
• message queuing if intended subscriber is down, resend message when it is up
• message delivery for one or some intended subscribers only

Maintainer

Eric Tang @ TYO LAB