'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, };