47 lines
2.0 KiB
JavaScript
47 lines
2.0 KiB
JavaScript
|
// Redis pub/sub backend — enables running MULTIPLE app instances. Each instance publishes every local
|
||
|
|
// real-time event; every other instance receives it and delivers to ITS local sockets. Selected by
|
||
|
|
// PUBSUB_BACKEND=redis; connection from REDIS_URL (default redis://bizgaze-redis:6379).
|
||
|
|
//
|
||
|
|
// Self-echo guard: Redis delivers a publish to ALL subscribers including the publisher, but the publishing
|
||
|
|
// instance ALREADY delivered locally — so every message is tagged with this instance's id and ignored on
|
||
|
|
// the way back in. Subscriptions made before connect are buffered and flushed in init().
|
||
|
|
const crypto = require('crypto');
|
||
|
|
const INSTANCE = crypto.randomBytes(8).toString('hex');
|
||
|
|
|
||
|
|
let pub = null, sub = null;
|
||
|
|
const pending = []; // [pattern, handler] queued before connect
|
||
|
|
|
||
|
|
async function doSubscribe(pattern, handler) {
|
||
|
|
const onMessage = (message, channel) => {
|
||
|
|
let m; try { m = JSON.parse(message); } catch { return; }
|
||
|
|
if (m.i === INSTANCE) return; // our own publish — already delivered locally
|
||
|
|
try { handler(channel, m.d); } catch (_) {}
|
||
|
|
};
|
||
|
|
if (pattern.includes('*')) await sub.pSubscribe(pattern, onMessage);
|
||
|
|
else await sub.subscribe(pattern, onMessage);
|
||
|
|
}
|
||
|
|
|
||
|
|
async function init() {
|
||
|
|
const { createClient } = require('redis');
|
||
|
|
const url = process.env.REDIS_URL || 'redis://bizgaze-redis:6379';
|
||
|
|
pub = createClient({ url });
|
||
|
|
sub = pub.duplicate();
|
||
|
|
pub.on('error', () => {}); sub.on('error', () => {}); // never let a redis blip crash the app
|
||
|
|
await pub.connect();
|
||
|
|
await sub.connect();
|
||
|
|
for (const [pattern, handler] of pending) { try { await doSubscribe(pattern, handler); } catch (_) {} }
|
||
|
|
pending.length = 0;
|
||
|
|
}
|
||
|
|
|
||
|
|
function publish(channel, data) {
|
||
|
|
if (!pub) return;
|
||
|
|
pub.publish(channel, JSON.stringify({ i: INSTANCE, d: data })).catch(() => {});
|
||
|
|
}
|
||
|
|
|
||
|
|
function subscribe(pattern, handler) {
|
||
|
|
if (!sub) { pending.push([pattern, handler]); return; } // buffer until init() connects
|
||
|
|
doSubscribe(pattern, handler).catch(() => {});
|
||
|
|
}
|
||
|
|
|
||
|
|
module.exports = { name: 'redis', init, publish, subscribe };
|