Vista Cues 1.0.0: which Vista 3 cue lists are live, from the network MIDI that fires them
Vista 3 reports nothing about what it is playing — its OSC only takes commands — so a booth with no lighting operator can't see the lighting computer. ProPresenter, the tracks rig and Companion fire Vista's cue lists with notes over macOS network MIDI sessions, and a session sends every note to every participant, so the module joins those sessions as one more listener (RTP-MIDI in plain Node, no helper) and hears what Vista hears. The admin's cue sheet turns notes into Go / release / pause / back / flash on named lists. Sheets with no release notes use [groups]: a Go replaces its group's live list, and the first group is the headline — the main scene — which takes over the look and clears the FX below it. The tile shows the headline list large with how long ago and who fired it, the other live lists as tags, recent notes, and which senders are connected. The module invites each configured sender and keeps re-inviting, accepts invitations from anyone, and advertises itself over Bonjour on macOS. A busy port (a config save re-initing before the old sockets close) is retried rather than failing. Default session port is 5040/5041, clear of the 5009 session on NA Ingest. tools/vista-midi-mock.js is a stand-in sender for development; modules/vista-cues/vista.test.js covers the parser, packet decoding and real loopback sessions. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,630 @@
|
||||
'use strict';
|
||||
|
||||
/**
|
||||
* A listen-only RTP-MIDI (AppleMIDI) session participant, in plain Node.
|
||||
*
|
||||
* This is the protocol behind a macOS "Network" MIDI session (Audio MIDI
|
||||
* Setup → MIDI Network Setup) and rtpMIDI on Windows: two UDP ports, a
|
||||
* control port P and a data port P+1. A session sends every MIDI message to
|
||||
* every participant it is connected to, so joining the same session Vista is
|
||||
* in hears exactly the notes Vista hears.
|
||||
*
|
||||
* Both ways of joining work:
|
||||
* - someone invites us (we show up in Audio MIDI Setup's Directory, or they
|
||||
* add our address by hand) and we accept anyone;
|
||||
* - we invite them (`invite(host, port)`), and keep re-inviting until they
|
||||
* accept, then keep the connection alive with clock syncs.
|
||||
*
|
||||
* We never send MIDI. Packet layouts follow Apple's AppleMIDI session
|
||||
* protocol and RFC 6295 (RTP payload for MIDI).
|
||||
*/
|
||||
|
||||
const dgram = require('dgram');
|
||||
const dns = require('dns');
|
||||
const crypto = require('crypto');
|
||||
const { EventEmitter } = require('events');
|
||||
|
||||
const SIG = 0xffff;
|
||||
const CMD = { IN: 0x494e, OK: 0x4f4b, NO: 0x4e4f, BY: 0x4259, CK: 0x434b, RS: 0x5253 };
|
||||
const PROTOCOL_VERSION = 2;
|
||||
|
||||
const INVITE_RETRY_MS = 1500; // between invitation attempts while setting up
|
||||
const INVITE_ATTEMPTS = 12; // then give up for a while…
|
||||
const REINVITE_MS = 5000; // …and start over after this long
|
||||
const SYNC_FAST_MS = 1500; // first few clock syncs, close together
|
||||
const SYNC_MS = 10000; // then this often, as macOS does
|
||||
const SYNC_TIMEOUT_MS = 25000; // no sync answer for this long → connection is gone
|
||||
const IDLE_MS = 90000; // an inbound participant we have not heard from at all
|
||||
const FEEDBACK_MS = 1000; // receiver feedback (lets the sender trim its journal)
|
||||
|
||||
/** The session clock: 100 µs ticks since this process started the session. */
|
||||
function makeClock() {
|
||||
const t0 = process.hrtime.bigint();
|
||||
return () => (process.hrtime.bigint() - t0) / 100000n;
|
||||
}
|
||||
|
||||
function randomU32() {
|
||||
return crypto.randomBytes(4).readUInt32BE(0);
|
||||
}
|
||||
|
||||
/* ── packets ─────────────────────────────────────────────────────────── */
|
||||
|
||||
function sessionPacket(cmd, token, ssrc, name) {
|
||||
const nameBuf = name == null ? Buffer.alloc(0) : Buffer.from(`${name}\0`, 'utf8');
|
||||
const b = Buffer.alloc(16 + nameBuf.length);
|
||||
b.writeUInt16BE(SIG, 0);
|
||||
b.writeUInt16BE(cmd, 2);
|
||||
b.writeUInt32BE(PROTOCOL_VERSION, 4);
|
||||
b.writeUInt32BE(token >>> 0, 8);
|
||||
b.writeUInt32BE(ssrc >>> 0, 12);
|
||||
nameBuf.copy(b, 16);
|
||||
return b;
|
||||
}
|
||||
|
||||
function syncPacket(ssrc, count, ts) {
|
||||
const b = Buffer.alloc(36);
|
||||
b.writeUInt16BE(SIG, 0);
|
||||
b.writeUInt16BE(CMD.CK, 2);
|
||||
b.writeUInt32BE(ssrc >>> 0, 4);
|
||||
b.writeUInt8(count, 8);
|
||||
for (let i = 0; i < 3; i++) b.writeBigUInt64BE(BigInt.asUintN(64, ts[i] || 0n), 12 + i * 8);
|
||||
return b;
|
||||
}
|
||||
|
||||
function feedbackPacket(ssrc, seq) {
|
||||
const b = Buffer.alloc(12);
|
||||
b.writeUInt16BE(SIG, 0);
|
||||
b.writeUInt16BE(CMD.RS, 2);
|
||||
b.writeUInt32BE(ssrc >>> 0, 4);
|
||||
b.writeUInt16BE(seq & 0xffff, 8);
|
||||
return b;
|
||||
}
|
||||
|
||||
/** A session command (IN/OK/NO/BY/CK/RS), or null when it isn't one. */
|
||||
function parseSession(buf) {
|
||||
if (buf.length < 4 || buf.readUInt16BE(0) !== SIG) return null;
|
||||
const cmd = buf.readUInt16BE(2);
|
||||
if (cmd === CMD.CK) {
|
||||
if (buf.length < 36) return null;
|
||||
return {
|
||||
cmd,
|
||||
ssrc: buf.readUInt32BE(4),
|
||||
count: buf.readUInt8(8),
|
||||
ts: [buf.readBigUInt64BE(12), buf.readBigUInt64BE(20), buf.readBigUInt64BE(28)],
|
||||
};
|
||||
}
|
||||
if (cmd === CMD.RS) {
|
||||
if (buf.length < 12) return null;
|
||||
return { cmd, ssrc: buf.readUInt32BE(4), seq: buf.readUInt16BE(8) };
|
||||
}
|
||||
if (cmd === CMD.IN || cmd === CMD.OK || cmd === CMD.NO || cmd === CMD.BY) {
|
||||
if (buf.length < 16) return null;
|
||||
let name = '';
|
||||
if (buf.length > 16) {
|
||||
const end = buf.indexOf(0, 16);
|
||||
name = buf.toString('utf8', 16, end < 0 ? buf.length : end);
|
||||
}
|
||||
return { cmd, version: buf.readUInt32BE(4), token: buf.readUInt32BE(8), ssrc: buf.readUInt32BE(12), name };
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/** Data bytes that follow a status byte (sysex is handled separately). */
|
||||
function dataLength(status) {
|
||||
const hi = status & 0xf0;
|
||||
if (hi === 0xc0 || hi === 0xd0) return 1;
|
||||
if (hi < 0xf0) return 2;
|
||||
if (status === 0xf1 || status === 0xf3) return 1;
|
||||
if (status === 0xf2) return 2;
|
||||
return 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* The MIDI commands in one RTP-MIDI packet (RFC 6295 §3), as
|
||||
* `{ status, data: [..] }` in order. Journal data is ignored: we only show
|
||||
* what is happening, and a note lost on the wire is one Vista also missed.
|
||||
*/
|
||||
function parseRtpMidi(buf) {
|
||||
if (buf.length < 13 || (buf[0] >> 6) !== 2) return null;
|
||||
const cc = buf[0] & 0x0f;
|
||||
const ext = Boolean(buf[0] & 0x10);
|
||||
const seq = buf.readUInt16BE(2);
|
||||
const ssrc = buf.readUInt32BE(8);
|
||||
let off = 12 + cc * 4;
|
||||
if (ext) {
|
||||
if (buf.length < off + 4) return null;
|
||||
off += 4 + buf.readUInt16BE(off + 2) * 4;
|
||||
}
|
||||
if (off >= buf.length) return null;
|
||||
const header = buf[off];
|
||||
const big = Boolean(header & 0x80);
|
||||
const zero = Boolean(header & 0x20);
|
||||
let len = header & 0x0f;
|
||||
if (big) {
|
||||
if (off + 1 >= buf.length) return null;
|
||||
len = (len << 8) | buf[off + 1];
|
||||
off += 2;
|
||||
} else {
|
||||
off += 1;
|
||||
}
|
||||
const end = Math.min(buf.length, off + len);
|
||||
const messages = [];
|
||||
let running = 0;
|
||||
let first = true;
|
||||
while (off < end) {
|
||||
if (!first || zero) {
|
||||
// Delta time: 1–4 octets, high bit means "more follow".
|
||||
let n = 0;
|
||||
while (off < end && (buf[off] & 0x80) && n < 3) { off++; n++; }
|
||||
off++;
|
||||
if (off >= end) break;
|
||||
}
|
||||
first = false;
|
||||
let status = buf[off];
|
||||
if (status & 0x80) {
|
||||
off++;
|
||||
} else if (running) {
|
||||
status = running;
|
||||
} else {
|
||||
break; // data byte with nothing to run on: give up on this packet
|
||||
}
|
||||
if (status === 0xf0 || status === 0xf7) {
|
||||
// A sysex, or a segment of one: runs to F7 (end), F0 (more follows) or F4 (cancelled).
|
||||
const start = off;
|
||||
while (off < end && buf[off] !== 0xf7 && buf[off] !== 0xf0 && buf[off] !== 0xf4) off++;
|
||||
const data = Array.from(buf.subarray(start, off));
|
||||
off++;
|
||||
messages.push({ status, data });
|
||||
running = 0;
|
||||
continue;
|
||||
}
|
||||
if (status < 0xf0) running = status;
|
||||
else if (status < 0xf8) running = 0; // system common cancels running status; realtime doesn't
|
||||
const n = dataLength(status);
|
||||
if (off + n > end) break;
|
||||
messages.push({ status, data: Array.from(buf.subarray(off, off + n)) });
|
||||
off += n;
|
||||
}
|
||||
return { seq, ssrc, messages };
|
||||
}
|
||||
|
||||
/** `{ status, data }` → a plain event, or null for what nobody here cares about. */
|
||||
function describe(msg) {
|
||||
const hi = msg.status & 0xf0;
|
||||
const channel = (msg.status & 0x0f) + 1;
|
||||
if (hi === 0x90 && msg.data[1] > 0) return { type: 'noteon', channel, note: msg.data[0], velocity: msg.data[1] };
|
||||
if (hi === 0x80 || (hi === 0x90 && msg.data[1] === 0)) return { type: 'noteoff', channel, note: msg.data[0] };
|
||||
if (hi === 0xc0) return { type: 'program', channel, program: msg.data[0] };
|
||||
if (hi === 0xb0) return { type: 'cc', channel, controller: msg.data[0], value: msg.data[1] };
|
||||
return null;
|
||||
}
|
||||
|
||||
/* ── the session ─────────────────────────────────────────────────────── */
|
||||
|
||||
class Session extends EventEmitter {
|
||||
/**
|
||||
* @param {{ name: string, port: number, log?: Function }} opts
|
||||
* Events: 'midi' ({ participant, event }), 'change' (participants or
|
||||
* listening changed), 'listening' (both ports bound).
|
||||
*/
|
||||
constructor({ name, port, log }) {
|
||||
super();
|
||||
this.name = name || 'ProdDash';
|
||||
this.port = port;
|
||||
this.log = log || (() => {});
|
||||
this.ssrc = randomU32();
|
||||
this.now = makeClock();
|
||||
this.control = null;
|
||||
this.data = null;
|
||||
this.participants = new Map(); // key → participant
|
||||
this.invites = new Map(); // key "host:port" → outbound invitation
|
||||
this.strangers = new Map(); // address → when we last told it we're not in its session
|
||||
this.timers = new Set();
|
||||
this.stopped = false;
|
||||
this.error = '';
|
||||
}
|
||||
|
||||
/**
|
||||
* Open both ports. Resolves once both are bound — or once the first try
|
||||
* failed, with `error` set. A port that is busy (our own previous session
|
||||
* still closing after a config save, or another program) is retried every
|
||||
* few seconds for as long as the session runs; 'listening' fires when it
|
||||
* finally binds.
|
||||
*/
|
||||
start() {
|
||||
this.every(FEEDBACK_MS, () => this.sendFeedback());
|
||||
this.every(5000, () => this.reap());
|
||||
return new Promise((resolve) => {
|
||||
let first = true;
|
||||
const attempt = (delay) => this.later(delay, async () => {
|
||||
const ok = await this.bind();
|
||||
if (ok) {
|
||||
this.error = '';
|
||||
this.emit('listening');
|
||||
this.emit('change');
|
||||
} else {
|
||||
this.emit('change');
|
||||
attempt(first ? 250 : 3000);
|
||||
}
|
||||
if (first) { first = false; resolve(ok); }
|
||||
});
|
||||
attempt(0);
|
||||
});
|
||||
}
|
||||
|
||||
/** Bind control and data sockets together; on failure close both. */
|
||||
bind() {
|
||||
const open = (port, which) => new Promise((resolve) => {
|
||||
const sock = dgram.createSocket('udp4');
|
||||
const fail = (e) => {
|
||||
try { sock.close(); } catch { /* not open */ }
|
||||
resolve({ sock: null, port, e });
|
||||
};
|
||||
sock.once('error', fail);
|
||||
sock.bind(port, () => {
|
||||
sock.removeListener('error', fail);
|
||||
sock.on('error', (e) => this.log(`socket error on ${port}: ${e.message}`));
|
||||
sock.on('message', (buf, rinfo) => {
|
||||
try { this.onPacket(buf, rinfo, which); } catch (e) { this.log(`bad packet from ${rinfo.address}: ${e.message}`); }
|
||||
});
|
||||
resolve({ sock, port });
|
||||
});
|
||||
});
|
||||
return Promise.all([open(this.port, 'control'), open(this.port + 1, 'data')]).then(([c, d]) => {
|
||||
if (this.stopped || !c.sock || !d.sock) {
|
||||
for (const x of [c, d]) if (x.sock) { try { x.sock.close(); } catch { /* closed */ } }
|
||||
const bad = !c.sock ? c : d;
|
||||
if (bad.e) {
|
||||
const msg = bad.e.code === 'EADDRINUSE'
|
||||
? `UDP port ${bad.port} is in use on this machine — pick another session port in /admin`
|
||||
: `UDP port ${bad.port}: ${bad.e.message}`;
|
||||
if (msg !== this.error) this.log(msg);
|
||||
this.error = msg;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
this.control = c.sock;
|
||||
this.data = d.sock;
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
stop() {
|
||||
this.stopped = true;
|
||||
for (const t of this.timers) { clearTimeout(t); clearInterval(t); }
|
||||
this.timers.clear();
|
||||
const control = this.control;
|
||||
const data = this.data;
|
||||
const close = () => {
|
||||
for (const s of [control, data]) { try { s && s.close(); } catch { /* closed */ } }
|
||||
};
|
||||
// Say goodbye, then close once the goodbyes have left (a config save
|
||||
// re-inits straight away and wants these ports back).
|
||||
const byes = [...this.participants.values()].filter((p) => p.connected);
|
||||
let left = byes.length;
|
||||
if (!control || !left) close();
|
||||
else {
|
||||
for (const p of byes) {
|
||||
try {
|
||||
control.send(sessionPacket(CMD.BY, p.token, this.ssrc), p.controlPort, p.address, () => { if (--left === 0) close(); });
|
||||
} catch { if (--left === 0) close(); }
|
||||
}
|
||||
}
|
||||
this.control = null;
|
||||
this.data = null;
|
||||
this.participants.clear();
|
||||
this.invites.clear();
|
||||
}
|
||||
|
||||
every(ms, fn) {
|
||||
const t = setInterval(fn, ms);
|
||||
t.unref?.();
|
||||
this.timers.add(t);
|
||||
return t;
|
||||
}
|
||||
|
||||
later(ms, fn) {
|
||||
const t = setTimeout(() => { this.timers.delete(t); if (!this.stopped) fn(); }, ms);
|
||||
t.unref?.();
|
||||
this.timers.add(t);
|
||||
return t;
|
||||
}
|
||||
|
||||
send(sock, buf, port, address) {
|
||||
if (!sock || this.stopped) return;
|
||||
try { sock.send(buf, port, address); } catch { /* socket closed */ }
|
||||
}
|
||||
|
||||
/** Every participant, for display. */
|
||||
list() {
|
||||
return [...this.participants.values()].map((p) => ({
|
||||
name: p.name,
|
||||
address: p.address,
|
||||
port: p.controlPort,
|
||||
connected: p.connected,
|
||||
outbound: p.outbound,
|
||||
lastHeard: p.lastHeard || null,
|
||||
since: p.since || null,
|
||||
}));
|
||||
}
|
||||
|
||||
/* — outbound invitations — */
|
||||
|
||||
/**
|
||||
* Keep a connection to the session at host:port. Re-invites on its own
|
||||
* whenever the connection drops. Returns the invitation's state object.
|
||||
*/
|
||||
invite(host, port, label) {
|
||||
const key = `${host}:${port}`;
|
||||
if (this.invites.has(key)) return this.invites.get(key);
|
||||
const inv = { key, host, port, label: label || host, address: '', token: 0, attempts: 0, stage: 'idle', error: '', timer: null };
|
||||
this.invites.set(key, inv);
|
||||
this.beginInvite(inv);
|
||||
return inv;
|
||||
}
|
||||
|
||||
beginInvite(inv) {
|
||||
if (this.stopped || !this.invites.has(inv.key)) return;
|
||||
dns.lookup(inv.host, { family: 4 }, (err, address) => {
|
||||
if (this.stopped) return;
|
||||
if (err) {
|
||||
inv.error = `can't find ${inv.host}`;
|
||||
inv.stage = 'idle';
|
||||
this.emit('change');
|
||||
this.later(REINVITE_MS, () => this.beginInvite(inv));
|
||||
return;
|
||||
}
|
||||
inv.address = address;
|
||||
// Already in a session with that machine (it invited us)? Nothing to do.
|
||||
const existing = this.findByAddress(address, inv.port);
|
||||
if (existing && existing.connected) {
|
||||
existing.label = inv.label;
|
||||
inv.stage = 'connected';
|
||||
inv.error = '';
|
||||
this.emit('change');
|
||||
return;
|
||||
}
|
||||
inv.token = randomU32();
|
||||
inv.attempts = 0;
|
||||
inv.stage = 'control';
|
||||
this.sendInvite(inv);
|
||||
});
|
||||
}
|
||||
|
||||
sendInvite(inv) {
|
||||
if (this.stopped || !this.invites.has(inv.key)) return;
|
||||
if (inv.stage !== 'control' && inv.stage !== 'data') return;
|
||||
if (inv.attempts >= INVITE_ATTEMPTS) {
|
||||
inv.error = inv.error || 'no answer';
|
||||
inv.stage = 'idle';
|
||||
this.emit('change');
|
||||
this.later(REINVITE_MS, () => this.beginInvite(inv));
|
||||
return;
|
||||
}
|
||||
inv.attempts++;
|
||||
const sock = inv.stage === 'control' ? this.control : this.data;
|
||||
const port = inv.stage === 'control' ? inv.port : inv.port + 1;
|
||||
this.send(sock, sessionPacket(CMD.IN, inv.token, this.ssrc, this.name), port, inv.address);
|
||||
clearTimeout(inv.timer);
|
||||
inv.timer = this.later(INVITE_RETRY_MS, () => this.sendInvite(inv));
|
||||
}
|
||||
|
||||
inviteFor(rinfo, token) {
|
||||
for (const inv of this.invites.values()) {
|
||||
if (inv.token === token && inv.address === rinfo.address) return inv;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
findByAddress(address, controlPort) {
|
||||
for (const p of this.participants.values()) {
|
||||
if (p.address === address && (controlPort == null || p.controlPort === controlPort)) return p;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/* — packets in — */
|
||||
|
||||
onPacket(buf, rinfo, which) {
|
||||
const s = parseSession(buf);
|
||||
if (!s) {
|
||||
if (which === 'data') this.onRtp(buf, rinfo);
|
||||
return;
|
||||
}
|
||||
const sock = which === 'control' ? this.control : this.data;
|
||||
switch (s.cmd) {
|
||||
case CMD.IN: return this.onInvitation(s, rinfo, which, sock);
|
||||
case CMD.OK: return this.onAccepted(s, rinfo, which);
|
||||
case CMD.NO: return this.onRejected(s, rinfo);
|
||||
case CMD.BY: return this.onBye(s, rinfo);
|
||||
case CMD.CK: return this.onSync(s, rinfo);
|
||||
default: return undefined; // RS from a peer: nothing for a listener to do
|
||||
}
|
||||
}
|
||||
|
||||
onInvitation(s, rinfo, which, sock) {
|
||||
// We accept anyone: being in more sessions only means hearing more cues.
|
||||
this.send(sock, sessionPacket(CMD.OK, s.token, this.ssrc, this.name), rinfo.port, rinfo.address);
|
||||
const key = `in:${s.ssrc}`;
|
||||
let p = this.participants.get(key);
|
||||
if (which === 'control') {
|
||||
if (!p) {
|
||||
p = {
|
||||
key, ssrc: s.ssrc, token: s.token, name: s.name || rinfo.address, label: '',
|
||||
address: rinfo.address, controlPort: rinfo.port, dataPort: rinfo.port + 1,
|
||||
outbound: false, connected: false, lastSeq: -1, fedBack: -1, lastHeard: Date.now(), since: 0,
|
||||
};
|
||||
this.participants.set(key, p);
|
||||
this.log(`invited by "${p.name}" at ${rinfo.address}:${rinfo.port}`);
|
||||
} else {
|
||||
p.token = s.token;
|
||||
p.controlPort = rinfo.port;
|
||||
p.name = s.name || p.name;
|
||||
}
|
||||
} else if (p) {
|
||||
p.dataPort = rinfo.port;
|
||||
p.connected = true;
|
||||
p.since = Date.now();
|
||||
p.lastHeard = Date.now();
|
||||
this.log(`in session with "${p.name}" (${p.address})`);
|
||||
for (const inv of this.invites.values()) {
|
||||
if (inv.address === p.address && inv.port === p.controlPort) {
|
||||
p.label = inv.label;
|
||||
inv.stage = 'connected';
|
||||
inv.error = '';
|
||||
clearTimeout(inv.timer);
|
||||
}
|
||||
}
|
||||
this.emit('change');
|
||||
}
|
||||
}
|
||||
|
||||
onAccepted(s, rinfo, which) {
|
||||
const inv = this.inviteFor(rinfo, s.token);
|
||||
if (!inv) return;
|
||||
clearTimeout(inv.timer);
|
||||
if (which === 'control' && inv.stage === 'control') {
|
||||
inv.stage = 'data';
|
||||
inv.attempts = 0;
|
||||
inv.remoteName = s.name;
|
||||
this.sendInvite(inv);
|
||||
return;
|
||||
}
|
||||
if (which === 'data' && inv.stage === 'data') {
|
||||
const key = `out:${inv.key}`;
|
||||
// Drop any stale record of the same remote before taking the new one.
|
||||
for (const [k, q] of this.participants) if (q.ssrc === s.ssrc && k !== key) this.participants.delete(k);
|
||||
const p = {
|
||||
key, ssrc: s.ssrc, token: inv.token, name: inv.remoteName || s.name || inv.host, label: inv.label,
|
||||
address: rinfo.address, controlPort: inv.port, dataPort: rinfo.port,
|
||||
outbound: true, connected: true, lastSeq: -1, fedBack: -1,
|
||||
lastHeard: Date.now(), lastSyncAnswer: Date.now(), since: Date.now(), syncs: 0,
|
||||
};
|
||||
this.participants.set(key, p);
|
||||
inv.stage = 'connected';
|
||||
inv.error = '';
|
||||
this.log(`joined "${p.name}" at ${inv.host} (${rinfo.address}:${inv.port})`);
|
||||
this.emit('change');
|
||||
this.startSync(p, inv);
|
||||
}
|
||||
}
|
||||
|
||||
onRejected(s, rinfo) {
|
||||
const inv = this.inviteFor(rinfo, s.token);
|
||||
if (!inv) return;
|
||||
clearTimeout(inv.timer);
|
||||
inv.stage = 'idle';
|
||||
inv.error = 'the session refused us — set "Who may connect" to Anyone in Audio MIDI Setup';
|
||||
this.log(`${inv.host}: invitation refused`);
|
||||
this.emit('change');
|
||||
this.later(REINVITE_MS * 2, () => this.beginInvite(inv));
|
||||
}
|
||||
|
||||
onBye(s, rinfo) {
|
||||
for (const [k, p] of this.participants) {
|
||||
if (p.ssrc === s.ssrc && p.address === rinfo.address) this.drop(k, 'left the session');
|
||||
}
|
||||
}
|
||||
|
||||
drop(key, why) {
|
||||
const p = this.participants.get(key);
|
||||
if (!p) return;
|
||||
this.participants.delete(key);
|
||||
clearInterval(p.syncTimer);
|
||||
this.timers.delete(p.syncTimer);
|
||||
this.log(`"${p.name}" ${why}`);
|
||||
// If we were asked to keep this one, go and get it back.
|
||||
for (const inv of this.invites.values()) {
|
||||
if (inv.address === p.address && inv.port === p.controlPort && inv.stage === 'connected') {
|
||||
inv.stage = 'idle';
|
||||
inv.error = why;
|
||||
this.later(p.outbound ? 1000 : REINVITE_MS, () => this.beginInvite(inv));
|
||||
}
|
||||
}
|
||||
this.emit('change');
|
||||
}
|
||||
|
||||
/* — clock sync — */
|
||||
|
||||
startSync(p) {
|
||||
const tick = () => {
|
||||
if (!this.participants.has(p.key)) return;
|
||||
if (Date.now() - p.lastSyncAnswer > SYNC_TIMEOUT_MS) {
|
||||
this.send(this.control, sessionPacket(CMD.BY, p.token, this.ssrc), p.controlPort, p.address);
|
||||
this.drop(p.key, 'stopped answering');
|
||||
return;
|
||||
}
|
||||
this.send(this.data, syncPacket(this.ssrc, 0, [this.now(), 0n, 0n]), p.dataPort, p.address);
|
||||
p.syncs++;
|
||||
if (p.syncs === 6) {
|
||||
clearInterval(p.syncTimer);
|
||||
this.timers.delete(p.syncTimer);
|
||||
p.syncTimer = this.every(SYNC_MS, tick);
|
||||
}
|
||||
};
|
||||
p.syncTimer = this.every(SYNC_FAST_MS, tick);
|
||||
tick();
|
||||
}
|
||||
|
||||
onSync(s, rinfo) {
|
||||
const p = [...this.participants.values()].find((q) => q.ssrc === s.ssrc);
|
||||
if (p) p.lastHeard = Date.now();
|
||||
if (s.count === 0) {
|
||||
this.send(this.data, syncPacket(this.ssrc, 1, [s.ts[0], this.now(), 0n]), rinfo.port, rinfo.address);
|
||||
} else if (s.count === 1) {
|
||||
if (p) p.lastSyncAnswer = Date.now();
|
||||
this.send(this.data, syncPacket(this.ssrc, 2, [s.ts[0], s.ts[1], this.now()]), rinfo.port, rinfo.address);
|
||||
}
|
||||
// count 2 completes a sync someone else started; nothing to answer.
|
||||
}
|
||||
|
||||
/** Drop inbound participants that have gone quiet (their machine slept or left without a BY). */
|
||||
reap() {
|
||||
const now = Date.now();
|
||||
for (const [k, p] of this.participants) {
|
||||
if (!p.outbound && now - (p.lastHeard || 0) > IDLE_MS) this.drop(k, 'went quiet');
|
||||
}
|
||||
}
|
||||
|
||||
sendFeedback() {
|
||||
for (const p of this.participants.values()) {
|
||||
if (!p.connected || p.lastSeq < 0 || p.lastSeq === p.fedBack) continue;
|
||||
p.fedBack = p.lastSeq;
|
||||
this.send(this.control, feedbackPacket(this.ssrc, p.lastSeq), p.controlPort, p.address);
|
||||
}
|
||||
}
|
||||
|
||||
/* — MIDI — */
|
||||
|
||||
onRtp(buf, rinfo) {
|
||||
const pkt = parseRtpMidi(buf);
|
||||
if (!pkt) return;
|
||||
let p = [...this.participants.values()].find((q) => q.ssrc === pkt.ssrc && q.connected);
|
||||
if (!p) {
|
||||
// MIDI from someone not in a session with us (a stale session after our restart).
|
||||
// Ask them to start over so the session comes back properly — once in a while, not per note.
|
||||
const last = this.strangers.get(rinfo.address) || 0;
|
||||
if (Date.now() - last > 5000) {
|
||||
this.strangers.set(rinfo.address, Date.now());
|
||||
this.send(this.control, sessionPacket(CMD.BY, 0, this.ssrc), rinfo.port - 1, rinfo.address);
|
||||
}
|
||||
return;
|
||||
}
|
||||
p.lastHeard = Date.now();
|
||||
p.lastSeq = pkt.seq;
|
||||
for (const m of pkt.messages) {
|
||||
const event = describe(m);
|
||||
if (event) this.emit('midi', { participant: p, event });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
Session,
|
||||
parseRtpMidi,
|
||||
parseSession,
|
||||
describe,
|
||||
sessionPacket,
|
||||
syncPacket,
|
||||
CMD,
|
||||
};
|
||||
Reference in New Issue
Block a user