237 lines
12 KiB
JavaScript
237 lines
12 KiB
JavaScript
 
|
/*
|
|||
|
|
* follower-link.js — the relay that lets a second screen mirror the stage TV exactly.
|
|||
|
|
*
|
|||
|
|
* WHY THIS EXISTS: the booth needs the Bible TV's picture in ProPresenter (and on to the ATEM)
|
|||
|
|
* without running a video cable out of the stage PC and without pushing video over Wi-Fi. So no
|
|||
|
|
* pixels are sent at all. The stage renderer publishes STATE — which tab, which passage, the scroll
|
|||
|
|
* position, and every pen stroke as vectors — and a browser on the booth machine re-renders the same
|
|||
|
|
* page natively at full resolution and frame rate. A few KB/s of JSON replaces ~250 Mbit/s of video,
|
|||
|
|
* and a lost packet costs a stale frame rather than a torn picture.
|
|||
|
|
*
|
|||
|
|
* stage renderer (follower-publish.js) ──WS /follower-ws?role=publisher──▶ here
|
|||
|
|
* │ fan-out
|
|||
|
|
* booth browser (index.html?follower=1) ◀─WS /follower-ws?role=subscriber───┘
|
|||
|
|
*
|
|||
|
|
* Two message classes, deliberately separated:
|
|||
|
|
* - `snapshot` — the whole UI state (tab, passage, geometry, committed strokes). Retained, so a
|
|||
|
|
* follower that connects or reconnects mid-service is correct within one frame.
|
|||
|
|
* - everything else (`stroke:*`, `scroll`) — high-rate deltas, relayed and forgotten.
|
|||
|
|
*
|
|||
|
|
* The retained snapshot is the reason a follower can be power-cycled mid-sermon and come back right.
|
|||
|
|
*
|
|||
|
|
* Only ONE publisher is honoured at a time (the newest wins): two publishers would interleave
|
|||
|
|
* contradictory states, and on a stage that shows up as text flickering between two passages. A
|
|||
|
|
* second copy of the app opened by accident is therefore harmless — it simply takes over, and the
|
|||
|
|
* real TV's next snapshot (published every 250ms) takes it back.
|
|||
|
|
*/
|
|||
|
|
import { WebSocketServer } from 'ws';
|
|||
|
|
|
|||
|
|
const PATH = '/follower-ws';
|
|||
|
|
|
|||
|
|
export function attachFollowerLink(server, opts = {}) {
|
|||
|
|
const log = opts.log || ((...a) => console.log('[FOLLOWER]', ...a));
|
|||
|
|
// Settings › Follower Screen switch. Off = no screen may subscribe; anyone connected is dropped.
|
|||
|
|
const isEnabled = typeof opts.isEnabled === 'function' ? opts.isEnabled : () => true;
|
|||
|
|
const wss = new WebSocketServer({ noServer: true });
|
|||
|
|
|
|||
|
|
let publisher = null; // the stage renderer
|
|||
|
|
const subscribers = new Set(); // booth screens
|
|||
|
|
let snapshot = null; // last full state, replayed to anyone who joins
|
|||
|
|
let seq = 0;
|
|||
|
|
|
|||
|
|
const stats = { published: 0, relayed: 0, dropped: 0, subscribersPeak: 0, startedAt: Date.now() };
|
|||
|
|
|
|||
|
|
function sendTo(ws, obj) {
|
|||
|
|
// readyState 1 = OPEN. Sending to a socket that is closing throws and would kill the relay.
|
|||
|
|
if (!ws || ws.readyState !== 1) return false;
|
|||
|
|
try { ws.send(typeof obj === 'string' ? obj : JSON.stringify(obj)); return true; } catch (_) { return false; }
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
function fanOut(raw) {
|
|||
|
|
stats.relayed++;
|
|||
|
|
for (const ws of subscribers) {
|
|||
|
|
/*
|
|||
|
|
* Never queue behind a slow screen. A booth machine that stalls (or a Wi-Fi hiccup) would
|
|||
|
|
* otherwise build an unbounded backlog and then replay a burst of stale stroke points —
|
|||
|
|
* the line would visibly redraw itself late. Skipping while the buffer is deep keeps the
|
|||
|
|
* follower in the present; the next snapshot repairs whatever it missed.
|
|||
|
|
*/
|
|||
|
|
if (ws.bufferedAmount > 1 << 20) { stats.dropped++; ws._missed = true; continue; }
|
|||
|
|
/*
|
|||
|
|
* It missed something. A snapshot only CARRIES the stroke list when it changed, so a skipped
|
|||
|
|
* clear-all or a skipped restored list would otherwise never be repaired while the socket
|
|||
|
|
* stays open. Hand it the completed retained snapshot (which always carries the list) before
|
|||
|
|
* the current message.
|
|||
|
|
*/
|
|||
|
|
if (ws._missed) { ws._missed = false; if (snapshot) sendTo(ws, snapshot); }
|
|||
|
|
sendTo(ws, raw);
|
|||
|
|
}
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/*
|
|||
|
|
* Last known committed strokes per passage. A snapshot only CARRIES the stroke list when it changed
|
|||
|
|
* (bandwidth), so the retained snapshot handed to a late joiner could otherwise lack it — and a
|
|||
|
|
* follower that reconnected after a Wi-Fi blip would show the chapter with no ink until the next
|
|||
|
|
* stroke. The retained copy is completed from here before it is stored.
|
|||
|
|
*/
|
|||
|
|
const lastStrokes = new Map();
|
|||
|
|
function remember(msg) {
|
|||
|
|
if (msg && msg.passageKey && Array.isArray(msg.strokes)) lastStrokes.set(msg.passageKey, msg.strokes);
|
|||
|
|
}
|
|||
|
|
function complete(msg) {
|
|||
|
|
if (msg && msg.passageKey && !Array.isArray(msg.strokes) && lastStrokes.has(msg.passageKey)) {
|
|||
|
|
msg.strokes = lastStrokes.get(msg.passageKey);
|
|||
|
|
}
|
|||
|
|
return msg;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/*
|
|||
|
|
* Only the machine this server runs on may publish. The stage renderer connects to its own server
|
|||
|
|
* over loopback; anything arriving from a LAN address is some other copy of the app — a browser
|
|||
|
|
* peeking at the stage URL, a booth screen, the admin's laptop. Letting it publish would hijack the
|
|||
|
|
* program feed: two publishers alternate as each reconnects and the follower flips between their
|
|||
|
|
* states. (Exactly what happened once with a forgotten test tab.) Refused here regardless of what
|
|||
|
|
* the client asks for, so no client-side mistake can ever cause it again.
|
|||
|
|
*/
|
|||
|
|
function isLoopback(req) {
|
|||
|
|
const a = String((req.socket && req.socket.remoteAddress) || '');
|
|||
|
|
return a === '127.0.0.1' || a === '::1' || a === '::ffff:127.0.0.1' || a.startsWith('127.');
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
/*
|
|||
|
|
* Second line of defence, for clients this server did not ship: `stage=1` is added to the URL only
|
|||
|
|
* by the Electron stage window (follower-publish.js). A browser tab on the stage machine running an
|
|||
|
|
* OLD cached copy of the app is loopback AND still publishes (it predates the client-side gate) —
|
|||
|
|
* it did exactly that once, alternating with the real stage 229 times in eight minutes and making
|
|||
|
|
* the booth feed flip between two states. So: while a marked stage publisher is connected, every
|
|||
|
|
* unmarked publisher is refused; a marked one always displaces an unmarked one; and only another
|
|||
|
|
* marked publisher (a second Electron window — should not happen) can displace a marked one.
|
|||
|
|
*/
|
|||
|
|
let publisherIsStage = false;
|
|||
|
|
|
|||
|
|
wss.on('connection', (ws, req) => {
|
|||
|
|
let role = 'subscriber', marked = false;
|
|||
|
|
try {
|
|||
|
|
const q = new URL(req.url, 'http://x').searchParams;
|
|||
|
|
role = q.get('role') === 'publisher' ? 'publisher' : 'subscriber';
|
|||
|
|
marked = q.get('stage') === '1';
|
|||
|
|
} catch (_) {}
|
|||
|
|
|
|||
|
|
if (role === 'publisher' && !isLoopback(req)) {
|
|||
|
|
log(`refused publisher from ${req.socket && req.socket.remoteAddress} — only the stage machine may publish`);
|
|||
|
|
stats.refusedPublishers = (stats.refusedPublishers || 0) + 1;
|
|||
|
|
try { ws.close(4003, 'publisher must be local'); } catch (_) {}
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
if (role === 'publisher') {
|
|||
|
|
if (publisher && publisher !== ws) {
|
|||
|
|
if (publisherIsStage && !marked) {
|
|||
|
|
log('refused an unmarked local publisher (an old browser copy of the app?) — the stage window is already publishing');
|
|||
|
|
stats.refusedPublishers = (stats.refusedPublishers || 0) + 1;
|
|||
|
|
try { ws.close(4004, 'stage window is publishing'); } catch (_) {}
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
log(`WARNING: a ${marked ? 'stage' : 'local'} publisher replaced the previous ${publisherIsStage ? 'stage' : 'local'} one`);
|
|||
|
|
stats.publisherReplacements = (stats.publisherReplacements || 0) + 1;
|
|||
|
|
try { publisher.close(4000, 'replaced'); } catch (_) {}
|
|||
|
|
}
|
|||
|
|
publisher = ws;
|
|||
|
|
publisherIsStage = marked;
|
|||
|
|
// A new stage session: nothing the previous one published (its ink, its passage) may be handed to
|
|||
|
|
// a screen that joins now. The new publisher's first snapshot arrives within a tick.
|
|||
|
|
lastStrokes.clear();
|
|||
|
|
snapshot = null;
|
|||
|
|
log(`${marked ? 'stage renderer' : 'local publisher (unmarked)'} connected`);
|
|||
|
|
} else {
|
|||
|
|
if (!isEnabled()) {
|
|||
|
|
stats.refusedSubscribers = (stats.refusedSubscribers || 0) + 1;
|
|||
|
|
try { ws.close(4005, 'follower screen is turned off in Settings'); } catch (_) {}
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
subscribers.add(ws);
|
|||
|
|
stats.subscribersPeak = Math.max(stats.subscribersPeak, subscribers.size);
|
|||
|
|
log(`follower connected (${subscribers.size} watching)`);
|
|||
|
|
// Make the new screen correct immediately instead of at the next state change.
|
|||
|
|
if (snapshot) sendTo(ws, snapshot);
|
|||
|
|
else sendTo(ws, JSON.stringify({ type: 'waiting' }));
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
ws.on('message', (data, isBinary) => {
|
|||
|
|
if (isBinary || ws !== publisher) return; // only the stage renderer may publish
|
|||
|
|
const raw = data.toString();
|
|||
|
|
stats.published++;
|
|||
|
|
// Peek at the type without parsing the whole payload on the hot path: stroke points are
|
|||
|
|
// relayed byte-for-byte, and only a snapshot needs to be retained.
|
|||
|
|
if (raw.indexOf('"snapshot"') !== -1 || raw.indexOf('"commit"') !== -1 || raw.indexOf('"clearAll"') !== -1) {
|
|||
|
|
try {
|
|||
|
|
const msg = JSON.parse(raw);
|
|||
|
|
if (msg && msg.type === 'clearAll') {
|
|||
|
|
// The stage wiped every passage. Forget every remembered list, and strip the ink
|
|||
|
|
// from the retained snapshot so a screen joining right now gets none of the old marks.
|
|||
|
|
lastStrokes.clear();
|
|||
|
|
if (snapshot) { try { const s = JSON.parse(snapshot); s.strokes = []; snapshot = JSON.stringify(s); } catch (_) {} }
|
|||
|
|
fanOut(raw);
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
if (msg && msg.type === 'snapshot') {
|
|||
|
|
msg.seq = ++seq;
|
|||
|
|
remember(msg);
|
|||
|
|
// Live subscribers already hold the strokes; only the RETAINED copy is completed.
|
|||
|
|
fanOut(JSON.stringify(msg));
|
|||
|
|
snapshot = JSON.stringify(complete(msg));
|
|||
|
|
return;
|
|||
|
|
}
|
|||
|
|
if (msg && msg.type === 'commit') { remember(msg); fanOut(raw); return; }
|
|||
|
|
} catch (_) { /* fall through and relay as-is */ }
|
|||
|
|
}
|
|||
|
|
fanOut(raw);
|
|||
|
|
});
|
|||
|
|
|
|||
|
|
ws.on('close', () => {
|
|||
|
|
if (ws === publisher) { publisher = null; publisherIsStage = false; log('publisher disconnected'); }
|
|||
|
|
else if (subscribers.delete(ws)) log(`follower disconnected (${subscribers.size} watching)`);
|
|||
|
|
});
|
|||
|
|
ws.on('error', () => { try { ws.close(); } catch (_) {} });
|
|||
|
|
});
|
|||
|
|
|
|||
|
|
server.on('upgrade', (req, socket, head) => {
|
|||
|
|
let pathname = '';
|
|||
|
|
try { pathname = new URL(req.url, 'http://x').pathname; } catch (_) {}
|
|||
|
|
if (pathname !== PATH) return; // leave every other upgrade alone
|
|||
|
|
wss.handleUpgrade(req, socket, head, (ws) => wss.emit('connection', ws, req));
|
|||
|
|
});
|
|||
|
|
|
|||
|
|
/*
|
|||
|
|
* Heartbeat. A booth screen that loses Wi-Fi leaves a half-open socket that never fires 'close',
|
|||
|
|
* so without this the relay would keep "successfully" writing stroke points into a dead pipe and
|
|||
|
|
* report a follower that is not there.
|
|||
|
|
*/
|
|||
|
|
const beat = setInterval(() => {
|
|||
|
|
for (const ws of [publisher, ...subscribers]) { if (ws) { try { ws.ping(); } catch (_) {} } }
|
|||
|
|
}, 15000);
|
|||
|
|
try { beat.unref(); } catch (_) {}
|
|||
|
|
|
|||
|
|
return {
|
|||
|
|
// Called when the operator flips the Settings switch. Turning it off drops every screen at once,
|
|||
|
|
// so the program feed goes black immediately rather than at the next reconnect.
|
|||
|
|
applyEnabled() {
|
|||
|
|
if (isEnabled()) return;
|
|||
|
|
for (const ws of subscribers) { try { ws.close(4005, 'follower screen is turned off in Settings'); } catch (_) {} }
|
|||
|
|
subscribers.clear();
|
|||
|
|
},
|
|||
|
|
status() {
|
|||
|
|
return {
|
|||
|
|
enabled: isEnabled(),
|
|||
|
|
publisher: !!publisher,
|
|||
|
|
publisherIsStage,
|
|||
|
|
followers: subscribers.size,
|
|||
|
|
hasSnapshot: !!snapshot,
|
|||
|
|
seq,
|
|||
|
|
...stats,
|
|||
|
|
uptimeMs: Date.now() - stats.startedAt
|
|||
|
|
};
|
|||
|
|
}
|
|||
|
|
};
|
|||
|
|
}
|