import { InstanceBase, InstanceStatus, Regex } from '@companion-module/base' import { ConsoleLink } from './console-link.js' import { Relay } from './relay.js' import { decode, describe, foldPattern } from './osc.js' import { State } from './state.js' import { FADE_CURVES, GEQ_RANGE, LEVEL, PARAMS, K, addr, adjustedLevel, clamp, fadeLevelAt, indexProblem, meterId, meterSubscription, outputMeterId, outputMeterSubscription, parse, query, sameValue } from './protocol.js' import { getActionDefinitions } from './definitions/actions.js' import { getFeedbackDefinitions, feedbackIdsForEvent, watchKey, watchKeyForEvent } from './definitions/feedbacks.js' import { getVariableDefinitions, getAllVariableValues, getEventVariableValues, describeChange } from './definitions/variables.js' import { getPresetDefinitions } from './definitions/presets.js' const LEARN_CAP = 400 /** A fade writes a new level this often (20 steps per second at most) and skips steps that would change the level by less than FADE_MIN_DB. */ const FADE_STEP_MS = 50 const FADE_MIN_DB = 0.05 const FADE_MAX_MS = 60000 /** Timings in ms. An instance copies these to `this.timing` so tests can shorten them. */ const TIMING = { keepaliveMs: 5000, // /Console/Name/? ping silenceMs: 20000, // no traffic for this long = disconnected txPaceMs: 20, // paced query queue settleMs: 400, // wait after the queue drains for stragglers rebuildDebounceMs: 750, // definitions rebuild after names / buss modes / snapshots change replyWaitMs: 1000, // how long an action waits for a value it needs relazyMs: 5000, // an unanswered lazy load may be requested again after this long writeGuardMs: 400, // after a write, differing inbound values for that address are ignored; then one confirm query flushMs: 40, // coalesce variable and feedback updates meterThrottleMs: 100, // meter feedback checks, at most 10 Hz resubscribeMs: 500, // after an iPad sends /Meters/clear peakWatchMs: 10500, // keep re-checking peak holds this long after a full-scale reading bindRetryMs: 10000, // retry a failed port bind (console link or relay) errorLogMs: 30000, // at most one log line per error code + destination in this time } export const DEFAULTS = { host: '', consolePort: 8000, localPort: 9000, relayEnable: true, relayPort: 8000, relayClientPort: 9000, relayTimeout: 3600, debugLog: false, } /** An integer in [lo, hi] from a number or a numeric string, else null. */ function wholeNumber(v, lo, hi) { const n = typeof v === 'string' && v.trim() !== '' ? Number(v) : v return Number.isInteger(n) && n >= lo && n <= hi ? n : null } /** * Coerce and validate a raw config: { cfg, error }. cfg has numbers and booleans in their final form; when * something is invalid `error` is a clear message and cfg is null. Never throws. */ export function normalizeConfig(raw) { const c = { ...DEFAULTS, ...(raw && typeof raw === 'object' ? raw : {}) } const host = typeof c.host === 'string' ? c.host.trim() : '' if (!host) return { cfg: null, error: 'Console IP missing' } const consolePort = wholeNumber(c.consolePort, 1, 65535) if (consolePort === null) return { cfg: null, error: 'Console receive port must be a whole number from 1 to 65535' } const localPort = wholeNumber(c.localPort, 1, 65535) if (localPort === null) return { cfg: null, error: 'Local receive port must be a whole number from 1 to 65535' } const cfg = { host, consolePort, localPort, relayEnable: !!c.relayEnable, relayPort: DEFAULTS.relayPort, relayClientPort: DEFAULTS.relayClientPort, relayTimeout: DEFAULTS.relayTimeout, debugLog: !!c.debugLog } if (cfg.relayEnable) { const relayPort = wholeNumber(c.relayPort, 1, 65535) if (relayPort === null) return { cfg: null, error: 'Relay port must be a whole number from 1 to 65535' } const relayClientPort = wholeNumber(c.relayClientPort, 0, 65535) if (relayClientPort === null) return { cfg: null, error: 'iPad receive port must be a whole number from 0 to 65535' } const relayTimeout = wholeNumber(c.relayTimeout, 0, 86400) if (relayTimeout === null) return { cfg: null, error: 'The iPad idle timeout must be a whole number of seconds from 0 to 86400' } Object.assign(cfg, { relayPort, relayClientPort, relayTimeout }) } return { cfg, error: null } } // ---------- v1 -> v2 upgrade ---------- const plain = (o) => !!o && o.isExpression === false const fixed = (value) => ({ value, isExpression: false }) /** Copy the listed v1 options under their (possibly renamed) v2 keys; options that are missing stay missing. */ function carry(o, pairs) { const out = {} for (const [from, to] of pairs) if (o[from] !== undefined) out[to] = o[from] return out } /** The v2 action for a v1 one: { actionId, options }, or null when there is no equivalent (it is then left untouched). */ function mapV1Action(a, mon) { const o = a.options || {} switch (a.actionId) { case 'send_on': return 'send' in o ? { actionId: 'send_on', options: carry(o, [['channel', 'channel'], ['send', 'buss'], ['state', 'state']]) } : null case 'send_level': return 'send' in o ? { actionId: 'send_level_set', options: carry(o, [['channel', 'channel'], ['send', 'buss'], ['level', 'level']]) } : null case 'send_nudge': return 'send' in o ? { actionId: 'send_level_adjust', options: carry(o, [['channel', 'channel'], ['send', 'buss'], ['delta', 'delta']]) } : null case 'send_pan': return 'send' in o ? { actionId: 'send_pan_set', options: carry(o, [['channel', 'channel'], ['send', 'buss'], ['pan', 'pan']]) } : null case 'group_assign': return 'group' in o ? { actionId: 'group_assign', options: carry(o, [['channel', 'channel'], ['group', 'buss'], ['state', 'state']]) } : null case 'snapshot_fire': return 'number' in o ? { actionId: 'snapshot_fire', options: carry(o, [['number', 'snapshot']]) } : null case 'buss_mute': return plain(o.buss) && !('busses' in o) ? { actionId: 'buss_mute', options: { busses: fixed([o.buss.value]), ...carry(o, [['state', 'state']]) } } : null case 'buss_fader': return { actionId: 'buss_fader_set', options: carry(o, [['buss', 'buss'], ['level', 'level']]) } case 'mute': return plain(o.channel) ? { actionId: 'input_mute', options: { channels: fixed([o.channel.value]), ...carry(o, [['state', 'state']]) } } : null case 'fader': return { actionId: 'input_fader_set', options: carry(o, [['channel', 'channel'], ['level', 'level']]) } case 'resend': return { actionId: 'reload', options: {} } case 'mon_on': return mon.mode === 'group' ? { actionId: 'group_assign', options: { ...carry(o, [['channel', 'channel'], ['state', 'state']]), buss: fixed(mon.buss) } } : { actionId: 'send_on', options: { ...carry(o, [['channel', 'channel'], ['state', 'state']]), buss: fixed(mon.buss) } } case 'mon_level': return mon.mode === 'group' ? null : { actionId: 'send_level_set', options: { ...carry(o, [['channel', 'channel'], ['level', 'level']]), buss: fixed(mon.buss) } } case 'mon_nudge': return mon.mode === 'group' ? null : { actionId: 'send_level_adjust', options: { ...carry(o, [['channel', 'channel'], ['delta', 'delta']]), buss: fixed(mon.buss) } } default: return null } } function mapV1Feedback(f, mon) { const o = f.options || {} switch (f.feedbackId) { case 'send_on': return 'send' in o ? { feedbackId: 'send_on', options: carry(o, [['channel', 'channel'], ['send', 'buss']]) } : null case 'group_assigned': return 'group' in o ? { feedbackId: 'group_assigned', options: carry(o, [['channel', 'channel'], ['group', 'buss']]) } : null case 'muted': return { feedbackId: 'input_muted', options: carry(o, [['channel', 'channel']]) } case 'mon_on': return { feedbackId: mon.mode === 'group' ? 'group_assigned' : 'send_on', options: { ...carry(o, [['channel', 'channel']]), buss: fixed(mon.buss) } } case 'mon_level_above': return mon.mode === 'group' ? null : { feedbackId: 'send_level_compare', options: { ...carry(o, [['channel', 'channel'], ['threshold', 'threshold']]), buss: fixed(mon.buss), comparison: fixed('above') } } default: return null // buss_muted, connected and relay_clients keep their ids and option keys } } /** * v1 -> v2. Config: keep the connection settings, debugLog = the old verbose log flag, the old default idle * timeout (180 s) becomes the new default, everything else is dropped. Actions and feedbacks whose ids v2 reuses * with other option keys (and the single-buss "mon" ones) are mapped to their v2 form; the rest stay as they are. */ export function upgradeV1Config(_context, props) { const cfg = props.config const updatedConfig = cfg ? { host: cfg.host ?? DEFAULTS.host, consolePort: cfg.consolePort ?? DEFAULTS.consolePort, localPort: cfg.localPort ?? DEFAULTS.localPort, relayEnable: cfg.relayEnable ?? DEFAULTS.relayEnable, relayPort: cfg.relayPort ?? DEFAULTS.relayPort, relayClientPort: cfg.relayClientPort ?? DEFAULTS.relayClientPort, relayTimeout: cfg.relayTimeout === undefined || cfg.relayTimeout === 180 ? DEFAULTS.relayTimeout : cfg.relayTimeout, debugLog: cfg.debugLog ?? !!cfg.verboseLog, } : null const mon = { buss: Number(cfg?.monSend) || 24, mode: cfg?.monMode === 'group' ? 'group' : 'aux' } const updatedActions = (props.actions || []).map((a) => { const m = mapV1Action(a, mon) return m ? { ...a, actionId: m.actionId, options: m.options } : a }) const updatedFeedbacks = (props.feedbacks || []).map((f) => { const m = mapV1Feedback(f, mon) return m ? { ...f, feedbackId: m.feedbackId, options: m.options } : f }) return { updatedConfig, updatedActions, updatedFeedbacks } } export default class DigicoS31Instance extends InstanceBase { constructor(internal) { super(internal) this.timing = { ...TIMING } this.config = { ...DEFAULTS } this.cfg = null // the validated, normalised config the running link / relay were opened with this.opChain = Promise.resolve() // init / configUpdated / destroy never overlap this.state = new State() this.link = null this.relay = null this.relayClients = new Map() // iPad clients survive the relay being reopened this.keepaliveTimer = null this.linkRetryTimer = null this.relayRetryTimer = null this.txQueue = [] this.txTimer = null this.queued = new Set() this.idleCallbacks = [] this.timers = new Set() this.seen = new Set() this.requested = new Map() // lazily loaded pairs -> time requested, de-duplicated for a while this.waiters = new Map() // event key -> Set this.pendingWrites = new Map() // address -> { value, until, timer } this.fades = new Map() // address -> the running fade (one per address) this.fadeStarts = new Map() // address -> token of a fade still waiting for the level it starts from this.fadeSeq = 0 this.errorLog = new Map() // error code + destination -> time last logged this.connectedFlag = false this.statusName = 'connecting' this.statusText = '' this.relayProblem = null this.connectStartedAt = Date.now() this.reloadOnReconnect = false this.loading = false this.loadToken = 0 this.pendingVars = {} this.pendingFeedbacks = new Set() // feedback types (fallback) this.pendingIds = new Set() // feedback instance ids this.watchers = new Map() // watch key -> Set this.watching = new Map() // feedback instance id -> watch key this.flushTimer = null this.rebuildTimer = null this.rebuildPending = false this.lastDefs = {} // JSON of what was last pushed, per kind of definition this.definedVariables = new Set() this.presetStats = null this.meterWanted = new Map() // meter id -> { ent: 'input' | 'buss' | 'matrix', n, meter } this.meterDirtyIds = new Set() this.peakUntil = new Map() // meter id -> time until which its peak hold is re-checked this.meterTimer = null this.resubscribeTimer = null } // ---------- lifecycle ---------- /** Run init / configUpdated / destroy one after the other, whatever order Companion calls them in. */ run(fn) { const p = this.opChain.then(fn) this.opChain = p.catch(() => {}) return p } async init(config) { return this.run(async () => { this.config = { ...DEFAULTS, ...config } this.rebuildDefinitions() await this.connect() }) } /** Only what changed is touched: the console link, the relay, or just a flag / a live relay setting. */ async configUpdated(config) { return this.run(async () => { const before = this.cfg this.config = { ...DEFAULTS, ...config } const { cfg, error } = normalizeConfig(this.config) const linkChanged = !before || !cfg || before.host !== cfg.host || before.consolePort !== cfg.consolePort || before.localPort !== cfg.localPort const linkIdle = !this.link && !this.linkRetryTimer if (error || linkChanged || linkIdle) { await this.disconnect() this.state.reset() await this.connect() return } this.cfg = cfg if (before.relayEnable !== cfg.relayEnable || (cfg.relayEnable && before.relayPort !== cfg.relayPort)) { await this.reopenRelay() } else if (this.relay && cfg.relayEnable) { this.relay.update({ clientPort: cfg.relayClientPort, timeoutMs: cfg.relayTimeout * 1000 }) this.publishRelay() } // debugLog is read live from this.config }) } async destroy() { return this.run(async () => { await this.disconnect() for (const t of [this.flushTimer, this.rebuildTimer]) if (t) clearTimeout(t) this.flushTimer = null this.rebuildTimer = null }) } getConfigFields() { return [ { type: 'static-text', id: 'consoleInfo', width: 12, label: 'Console setup', value: 'On the console open Extensions > iPad. Set the iPad IP address to this computer, the Send Port to the "Local receive port" below, ' + 'the Receive Port to the "Console receive port" below, and enable the iPad connection. Companion then holds the console\'s single remote slot.', }, { type: 'textinput', id: 'host', label: 'Console IP', width: 4, default: DEFAULTS.host, regex: Regex.IP }, { type: 'number', id: 'consolePort', label: 'Console receive port (module sends here)', width: 4, default: DEFAULTS.consolePort, min: 1, max: 65535, asInteger: true }, { type: 'number', id: 'localPort', label: 'Local receive port (console sends here; = console "Send port")', width: 4, default: DEFAULTS.localPort, min: 1, max: 65535, asInteger: true, }, { type: 'static-text', id: 'ipadInfo', width: 12, label: 'iPad apps (relay)', value: 'In the DiGiCo S app on each iPad enter this computer as the console address, set the app\'s Send Port to the "Relay port" below and its Receive Port ' + 'to the "iPad receive port" below. Every app is relayed through this connection, so several iPads can be used at once.', }, { type: 'checkbox', id: 'relayEnable', label: 'Enable iPad relay', width: 3, default: DEFAULTS.relayEnable }, { type: 'number', id: 'relayPort', label: 'Relay port (iPad apps send here)', width: 3, default: DEFAULTS.relayPort, min: 1, max: 65535, asInteger: true }, { type: 'number', id: 'relayClientPort', label: 'iPad receive port (0 = reply to the app\'s source port)', width: 3, default: DEFAULTS.relayClientPort, min: 0, max: 65535, asInteger: true, }, { type: 'number', id: 'relayTimeout', label: 'Forget an iPad after (idle seconds, 0 = never)', width: 3, default: DEFAULTS.relayTimeout, min: 0, max: 86400, asInteger: true, }, { type: 'checkbox', id: 'debugLog', label: 'Debug logging (log every new address pattern and all traffic at debug level)', width: 12, default: DEFAULTS.debugLog, }, ] } /** Show the status that follows from the link state, with a relay problem as a warning while the link itself is fine. */ reportStatus() { switch (this.statusName) { case 'bad_config': this.updateStatus(InstanceStatus.BadConfig, this.statusText) break case 'failed': this.updateStatus(InstanceStatus.ConnectionFailure, this.statusText) break case 'disconnected': this.updateStatus(InstanceStatus.Disconnected, 'No reply from the console (check the iPad IP address and ports on the console)') break case 'ok': if (this.relayProblem) this.updateStatus(InstanceStatus.UnknownWarning, this.relayProblem) else this.updateStatus(InstanceStatus.Ok) break default: this.updateStatus(InstanceStatus.Connecting) } } /** One log line per error code + destination per `errorLogMs`, whatever the rate the errors come at. */ reportError(source, err) { const key = `${source}|${err.code || err.message}|${err.dest || ''}` const now = Date.now() const last = this.errorLog.get(key) if (last !== undefined && now - last < this.timing.errorLogMs) return false if (this.errorLog.size > 200) this.errorLog.clear() this.errorLog.set(key, now) // An iPad that went to sleep makes every fan-out fail with "host down/unreachable" until the relay forgets it // (up to the idle timeout): say so once per iPad as a warning instead of an error every few seconds. const asleep = source === 'Relay socket' && err.dest && ['EHOSTDOWN', 'EHOSTUNREACH', 'ENETUNREACH'].includes(err.code) if (asleep) { const ip = String(err.dest).split(':')[0] const sleepKey = `asleep|${ip}` const lastSleep = this.errorLog.get(sleepKey) if (lastSleep !== undefined && now - lastSleep < 10 * 60 * 1000) return false this.errorLog.set(sleepKey, now) this.log('warn', `Relay: iPad ${ip} is not reachable (asleep or off the network); it stays on the list until it sends again or the idle timeout passes`) return true } this.log('error', `${source}: ${err.message}${err.dest ? ` (${err.dest})` : ''}`) return true } async connect() { const { cfg, error } = normalizeConfig(this.config) this.cfg = cfg this.relayProblem = null if (error) { this.statusName = 'bad_config' this.statusText = error this.reportStatus() this.log('warn', error) return } this.statusName = 'connecting' this.reportStatus() this.seen.clear() this.requested.clear() this.meterWanted.clear() this.connectedFlag = false this.reloadOnReconnect = false this.connectStartedAt = Date.now() if (await this.openLink()) await this.afterLink() else this.scheduleLinkRetry() } async openLink() { const cfg = this.cfg const link = new ConsoleLink({ host: cfg.host, port: cfg.consolePort, localPort: cfg.localPort }) link.on('error', (err) => this.reportError('Console socket', err)) link.on('stray', (buf, rinfo) => this.learn('stray', decode(buf), rinfo)) link.on('raw', (buf) => { if (this.relay) this.relay.fanout(buf) }) link.on('message', (msg, _buf, rinfo) => this.onConsoleMessage(msg, rinfo)) try { const info = await link.open() this.link = link this.log('info', `Console link: listening on udp ${info.port}, sending to ${cfg.host}:${cfg.consolePort}`) return true } catch (err) { this.statusName = 'failed' this.statusText = `Cannot bind local port ${cfg.localPort}: ${err.code || err.message}` this.reportStatus() this.reportError('Console socket', Object.assign(err, { message: `cannot bind local port ${cfg.localPort}: ${err.message}` })) return false } } scheduleLinkRetry() { if (this.linkRetryTimer) clearTimeout(this.linkRetryTimer) this.linkRetryTimer = setTimeout(() => { this.linkRetryTimer = null this.run(async () => { if (this.link || !this.cfg) return if (await this.openLink()) { this.statusName = 'connecting' this.reportStatus() this.connectStartedAt = Date.now() await this.afterLink() } else this.scheduleLinkRetry() }) }, this.timing.bindRetryMs) } /** Everything that needs the console link: relay, first query, load, keepalive. */ async afterLink() { await this.openRelay() this.sendPing() this.startLoad('connect') this.keepaliveTimer = setInterval(() => this.tick(), this.timing.keepaliveMs) this.publishRelay() } async openRelay() { const cfg = this.cfg if (!cfg?.relayEnable) { this.relay = null this.relayProblem = null this.reportStatus() return } const relay = new Relay({ port: cfg.relayPort, clientPort: cfg.relayClientPort, timeoutMs: cfg.relayTimeout * 1000, clients: this.relayClients }) relay.on('error', (err) => this.reportError('Relay socket', err)) relay.on('client', (rinfo) => { this.log('info', `Relay: new iPad client ${rinfo.address}:${rinfo.port}`) this.publishRelay() }) relay.on('clientGone', (ip) => { this.log('info', `Relay: iPad client ${ip} forgotten`) this.publishRelay() }) relay.on('fromClient', (buf, rinfo) => { if (this.link) this.link.sendRaw(buf) const msg = decode(buf) // An iPad clears every meter subscription on the console when it connects, ours included. if (msg.address === addr.meterClear) this.scheduleMeterResubscribe() if (this.config.debugLog) { this.learn('ipad', msg, rinfo) this.log('debug', `ipad ${rinfo.address} -> console: ${describe(msg)}`) } }) try { const info = await relay.open() this.relay = relay this.relayProblem = null this.log('info', `Relay: iPad apps may send to udp ${info.port}; console traffic fans out to port ${cfg.relayClientPort || '(source)'}`) } catch (err) { this.relay = null this.relayProblem = err.code === 'EADDRINUSE' ? `iPad relay: port ${cfg.relayPort} in use` : `iPad relay: cannot bind port ${cfg.relayPort} (${err.code || err.message})` this.reportError('Relay socket', Object.assign(err, { message: `cannot bind port ${cfg.relayPort}: ${err.message}` })) this.scheduleRelayRetry() } this.reportStatus() } scheduleRelayRetry() { if (this.relayRetryTimer) clearTimeout(this.relayRetryTimer) this.relayRetryTimer = setTimeout(() => { this.relayRetryTimer = null this.run(async () => { if (!this.link || this.relay || !this.cfg?.relayEnable) return await this.openRelay() this.publishRelay() }) }, this.timing.bindRetryMs) } /** The relay port or enable flag changed: reopen it, keeping the iPad clients. */ async reopenRelay() { if (this.relayRetryTimer) clearTimeout(this.relayRetryTimer) this.relayRetryTimer = null if (this.relay) this.relay.close() this.relay = null this.relayProblem = null if (this.link) await this.openRelay() this.publishRelay() } async disconnect() { this.loadToken++ this.loading = false for (const t of [this.keepaliveTimer, this.txTimer, this.meterTimer]) if (t) clearInterval(t) for (const t of [this.linkRetryTimer, this.relayRetryTimer]) if (t) clearTimeout(t) this.keepaliveTimer = null this.linkRetryTimer = null this.relayRetryTimer = null this.txTimer = null this.meterTimer = null for (const t of this.timers) clearTimeout(t) this.timers.clear() this.pendingWrites.clear() this.fadeStarts.clear() if (this.fades.size) { this.fades.clear() // their step timers were cleared with the others above this.fadesChanged() } if (this.rebuildTimer) clearTimeout(this.rebuildTimer) this.rebuildTimer = null this.resubscribeTimer = null this.txQueue = [] this.queued.clear() this.idleCallbacks = [] if (this.link) this.link.close() if (this.relay) this.relay.close() this.link = null this.relay = null this.connectedFlag = false } later(fn, ms) { const t = setTimeout(() => { this.timers.delete(t) fn() }, ms) this.timers.add(t) return t } tick() { if (!this.link) return this.sendPing() const quietFor = Date.now() - Math.max(this.link.lastRx, this.connectStartedAt) if (quietFor > this.timing.silenceMs && this.statusName !== 'disconnected') this.setConnected(false) this.publishRelay() } setConnected(on) { const was = this.connectedFlag this.connectedFlag = !!on this.statusName = on ? 'ok' : 'disconnected' this.reportStatus() if (!on) this.reloadOnReconnect = true if (was === this.connectedFlag) return this.queueVariables({ connected: on ? 'Yes' : 'No' }) this.touch(watchKey.connected, ['connected']) this.log('info', on ? 'Console link alive' : 'Console link silent') if (on) { if (this.reloadOnReconnect && !this.loading) { this.reloadOnReconnect = false this.startLoad('reconnect') } this.resubscribeMeters() } } // ---------- outbound ---------- /** Refuse an address whose entity numbers are not whole numbers in range: one warning, no packet. */ addressOk(address) { const problem = indexProblem(address) if (!problem) return true this.log('warn', `Not sent (${problem}): ${address}`) return false } sendOsc(address, args = []) { if (!this.link) return false if (this.config.debugLog) this.log('debug', `-> ${address} ${JSON.stringify(args)}`) return this.link.send(address, args) } queryNow(address) { if (!this.link || !this.addressOk(address)) return false return this.link.send(`${address}/?`, []) } sendPing() { this.queryNow(addr.consoleName) } /** Paced, de-duplicated sends: bulk queries must not flood the console. */ enqueue(list) { if (!this.link) return for (const item of list) { const id = item[0] + (item[1].length ? JSON.stringify(item[1]) : '') if (this.queued.has(id)) continue this.queued.add(id) this.txQueue.push({ item, id }) } this.pump() } pump() { if (this.txTimer) return if (!this.txQueue.length) { this.fireIdle() return } this.txTimer = setInterval(() => { const next = this.txQueue.shift() if (!next) { clearInterval(this.txTimer) this.txTimer = null this.fireIdle() return } this.queued.delete(next.id) if (this.link) this.link.send(next.item[0], next.item[1]) }, this.timing.txPaceMs) } whenIdle(cb) { if (!this.txTimer && !this.txQueue.length) this.later(cb, 0) else this.idleCallbacks.push(cb) } fireIdle() { const cbs = this.idleCallbacks this.idleCallbacks = [] for (const cb of cbs) cb() } /** * Send a write now and apply it to the local state at once, so quick encoder turns accumulate. For * `writeGuardMs` after the last write to an address, a differing value arriving for it is stale (an echo of * an earlier click) and is ignored; when the guard runs out one query confirms the console's value. */ write(address, args, localValue) { if (!this.addressOk(address)) return false if (!this.link) { this.log('warn', `Not connected: ${address} not sent`) return false } this.sendOsc(address, args) const ev = this.state.applyMessage(address, [localValue]) if (ev) { this.setGuard(address, ev.value) this.afterEvent(ev) } return true } setGuard(address, value) { const old = this.pendingWrites.get(address) if (old) { clearTimeout(old.timer) this.timers.delete(old.timer) } const timer = this.later(() => { this.pendingWrites.delete(address) this.enqueue([query(address)]) }, this.timing.writeGuardMs) this.pendingWrites.set(address, { value, until: Date.now() + this.timing.writeGuardMs, timer }) } /** True when `msg` is a stale value for an address we wrote to a moment ago. */ isStale(msg) { const g = this.pendingWrites.get(msg.address) if (!g || Date.now() >= g.until) return false const incoming = parse(msg.address, msg.args) return !!incoming && incoming.value !== undefined && !sameValue(incoming.value, g.value) } waitFor(key, ms = this.timing.replyWaitMs) { return new Promise((resolve) => { let set = this.waiters.get(key) if (!set) { set = new Set() this.waiters.set(key, set) } const done = () => { clearTimeout(timer) set.delete(done) if (!set.size) this.waiters.delete(key) resolve() } const timer = setTimeout(done, ms) set.add(done) }) } notifyWaiters(key) { const set = this.waiters.get(key) if (set) for (const done of [...set]) done() } /** Current value, asking the console first (and waiting briefly for the reply) when it is not known yet. */ async current(key, get, address) { const v = get() if (v !== undefined) return v if (!this.addressOk(address)) return undefined const reply = this.waitFor(key) this.queryNow(address) await reply return get() } /** * Load one value pair on first use. Asked for again only after a while (an unanswered request may be * repeated, but not more often than every relazyMs); callers only ask while the value is unknown. */ ensure(key, addresses) { if (!this.link) return const now = Date.now() const last = this.requested.get(key) if (last !== undefined && now - last < Math.max(this.timing.replyWaitMs, this.timing.relazyMs)) return if (!addresses.every((a) => this.addressOk(a))) return this.requested.set(key, now) this.enqueue(addresses.map(query)) } ensureSend(ch, buss) { const s = this.state.getSend(ch, buss) if (s.level === undefined || s.on === undefined) this.ensure(`send:${ch}:${buss}`, [addr.sendLevel(ch, buss), addr.sendOn(ch, buss)]) } ensureSendPan(ch, buss) { if (this.state.getSend(ch, buss).pan === undefined) this.ensure(`pan:${ch}:${buss}`, [addr.sendPan(ch, buss)]) } ensureGroup(ch, buss) { if (this.state.getGroup(ch, buss) === undefined) this.ensure(`group:${ch}:${buss}`, [addr.groupAssign(ch, buss)]) } ensureMatrixSend(inN, outN) { const m = this.state.getMatrixSend(inN, outN) if (m.level === undefined || m.on === undefined) this.ensure(`msend:${inN}:${outN}`, [addr.matrixSendLevel(inN, outN), addr.matrixSendOn(inN, outN)]) } ensureGeqOn(n) { if (this.state.getGeq(n).on === undefined) this.ensure(`geq:${n}:on`, [addr.geqIn(n)]) } ensureGeqBand(n, band) { if (this.state.getGeq(n).gains?.get(Number(band)) === undefined) this.ensure(`geq:${n}:gain:${band}`, [addr.geqGain(n, band)]) } ensureParam(ent, n, param) { const address = addr.param(ent, n, this.sideOf(ent, n), param) if (address && this.state.getParam(ent, n, param) === undefined) this.ensure(`param:${ent}:${n}:${param}`, [address]) } // ---------- operations used by actions ---------- async writeFlag(address, key, get, mode) { if (!this.addressOk(address)) return false let on if (mode === 'on') on = true else if (mode === 'off') on = false else { const cur = await this.current(key, get, address) if (cur === undefined) { this.log('warn', `${address}: current state unknown (no reply from the console), toggle not applied`) return false } on = !cur } // Verified on an S31: switches are written as float 0.0 / 1.0; int and T/F writes are silently ignored. return this.write(address, [on ? 1 : 0], on ? 1 : 0) } /** Write a dB level, clamped to [lo, +10]; `lo` is -150 except for matrix sends (-90). A running fade on the address ends. */ writeLevel(address, dB, lo = LEVEL.min) { if (!this.addressOk(address)) return false this.cancelFade(address) const v = clamp(Number(dB), lo, LEVEL.max) if (!Number.isFinite(v)) return false return this.write(address, [v], v) } async adjustLevel(address, key, get, delta, lo = LEVEL.min) { if (!this.addressOk(address)) return false this.cancelFade(address) const cur = await this.current(key, get, address) if (cur === undefined) { this.log('warn', `${address}: current level unknown (no reply from the console), not adjusted`) return false } return this.writeLevel(address, adjustedLevel(cur, delta, lo), lo) } // ---------- fades ---------- /** * Move a level from where it is now to `targetDb` over `durationMs`: a new level every FADE_STEP_MS (20 Hz), each * a normal write(), the last one exactly the target. Curves are those of `fadeLevelAt`. There is one fade per * address: a new fade, a plain set or an adjust on the same address ends the running one, and closing the console * link ends them all. When the current level cannot be read, the target is written at once. */ async fadeLevel(address, key, get, targetDb, durationMs, curve = 'fader', lo = LEVEL.min) { if (!this.addressOk(address)) return false const target = clamp(Number(targetDb), lo, LEVEL.max) if (!Number.isFinite(target)) return false const ms = Math.min(Number(durationMs), FADE_MAX_MS) if (!(ms > 0)) return this.writeLevel(address, target, lo) this.cancelFade(address) const token = ++this.fadeSeq this.fadeStarts.set(address, token) const from = await this.current(key, get, address) if (this.fadeStarts.get(address) !== token) return false // another action took the address, fades were stopped or the link was closed meanwhile this.fadeStarts.delete(address) if (from === undefined || !Number.isFinite(from)) { this.log('warn', `${address}: current level unknown (no reply from the console), set without a fade`) return this.write(address, [target], target) } if (!this.link) return false const steps = Math.max(1, Math.ceil(ms / FADE_STEP_MS)) const fade = { address, from, target, curve: FADE_CURVES.includes(curve) ? curve : 'fader', lo, steps, interval: ms / steps, k: 0, last: from, startedAt: Date.now(), timer: null } this.fades.set(address, fade) this.fadesChanged() this.scheduleFadeStep(fade) return true } scheduleFadeStep(fade) { const due = fade.startedAt + Math.round((fade.k + 1) * fade.interval) fade.timer = this.later(() => this.fadeStep(fade), Math.max(0, due - Date.now())) } fadeStep(fade) { if (this.fades.get(fade.address) !== fade) return fade.k++ const final = fade.k >= fade.steps const db = fadeLevelAt(fade.curve, fade.from, fade.target, fade.k / fade.steps, fade.lo) if (final ? db !== fade.last : Math.abs(db - fade.last) >= FADE_MIN_DB) { if (!this.write(fade.address, [db], db)) { this.cancelFade(fade.address) return } fade.last = db } if (final) { this.fades.delete(fade.address) this.fadesChanged() } else { this.scheduleFadeStep(fade) } } /** End the fade running on `address` (or waiting for its starting level) where the level is now. */ cancelFade(address) { this.fadeStarts.delete(address) const fade = this.fades.get(address) if (!fade) return false clearTimeout(fade.timer) this.timers.delete(fade.timer) this.fades.delete(address) this.fadesChanged() return true } /** End every fade; each level stays where it is. */ stopFades() { this.fadeStarts.clear() if (!this.fades.size) return for (const fade of this.fades.values()) { clearTimeout(fade.timer) this.timers.delete(fade.timer) } this.fades.clear() this.fadesChanged() } fadesChanged() { this.queueVariables({ fades_running: this.fades.size }) this.touch(watchKey.fades, ['fade_running']) } writePan(address, pan) { if (!this.addressOk(address)) return false const v = clamp(Number(pan), 0, 1) if (!Number.isFinite(v)) return false return this.write(address, [v], v) } async adjustPanValue(address, key, get, pdelta) { if (!this.addressOk(address)) return false const cur = await this.current(key, get, address) if (cur === undefined) { this.log('warn', `${address}: current pan unknown (no reply from the console), not adjusted`) return false } return this.writePan(address, cur + Number(pdelta)) } sideOf(ent, n) { return ent === 'buss' ? this.state.bussMode(n) : undefined } setSwitch(ent, field, n, mode) { const side = this.sideOf(ent, n) const get = () => (field === 'mute' ? this.state.getMute(ent, n) : this.state.getSolo(ent, n)) return this.writeFlag(addr[field](ent, n, side), K.value(ent, n, field, side), get, mode) } /** Set a fader; with `durationMs` > 0 it moves there over that time (see fadeLevel). */ setFader(ent, n, dB, durationMs = 0, curve = 'fader') { const side = this.sideOf(ent, n) const address = addr.fader(ent, n, side) if (!(durationMs > 0)) return this.writeLevel(address, dB) return this.fadeLevel(address, K.value(ent, n, 'fader', side), () => this.state.getFader(ent, n), dB, durationMs, curve) } adjustFader(ent, n, delta) { const side = this.sideOf(ent, n) return this.adjustLevel(addr.fader(ent, n, side), K.value(ent, n, 'fader', side), () => this.state.getFader(ent, n), delta) } setPan(n, pan) { return this.writePan(addr.pan(n), pan) } adjustPan(n, pdelta) { return this.adjustPanValue(addr.pan(n), K.value('input', n, 'pan'), () => this.state.getPan(n), pdelta) } setInputName(n, name) { const text = String(name ?? '') if (!text.trim()) { this.log('warn', `Input ${n}: an empty name is ignored`) return false } return this.write(addr.name('input', n), [text], text) } setSendOn(ch, buss, mode) { return this.writeFlag(addr.sendOn(ch, buss), K.send(ch, buss, 'on'), () => this.state.getSend(ch, buss).on, mode) } setSendLevel(ch, buss, dB, durationMs = 0, curve = 'fader') { const address = addr.sendLevel(ch, buss) if (!(durationMs > 0)) return this.writeLevel(address, dB) return this.fadeLevel(address, K.send(ch, buss, 'level'), () => this.state.getSend(ch, buss).level, dB, durationMs, curve) } adjustSendLevel(ch, buss, delta) { return this.adjustLevel(addr.sendLevel(ch, buss), K.send(ch, buss, 'level'), () => this.state.getSend(ch, buss).level, delta) } setSendPan(ch, buss, pan) { return this.writePan(addr.sendPan(ch, buss), pan) } adjustSendPan(ch, buss, pdelta) { return this.adjustPanValue(addr.sendPan(ch, buss), K.send(ch, buss, 'pan'), () => this.state.getSend(ch, buss).pan, pdelta) } setGroup(ch, buss, mode) { return this.writeFlag(addr.groupAssign(ch, buss), K.group(ch, buss), () => this.state.getGroup(ch, buss), mode) } setMatrixSendOn(inN, outN, mode) { return this.writeFlag(addr.matrixSendOn(inN, outN), K.msend(inN, outN, 'on'), () => this.state.getMatrixSend(inN, outN).on, mode) } setMatrixSendLevel(inN, outN, dB, durationMs = 0, curve = 'fader') { const address = addr.matrixSendLevel(inN, outN) if (!(durationMs > 0)) return this.writeLevel(address, dB, LEVEL.matrixSendMin) return this.fadeLevel(address, K.msend(inN, outN, 'level'), () => this.state.getMatrixSend(inN, outN).level, dB, durationMs, curve, LEVEL.matrixSendMin) } adjustMatrixSendLevel(inN, outN, delta) { return this.adjustLevel(addr.matrixSendLevel(inN, outN), K.msend(inN, outN, 'level'), () => this.state.getMatrixSend(inN, outN).level, delta, LEVEL.matrixSendMin) } setGeqOn(n, mode) { return this.writeFlag(addr.geqIn(n), K.geq(n, 'on'), () => this.state.getGeq(n).on, mode) } setGeqBand(n, band, gain) { const address = addr.geqGain(n, band) if (!this.addressOk(address)) return false const v = clamp(Number(gain), GEQ_RANGE.min, GEQ_RANGE.max) if (!Number.isFinite(v)) return false return this.write(address, [v], v) } async adjustGeqBand(n, band, delta) { const address = addr.geqGain(n, band) if (!this.addressOk(address)) return false const cur = await this.current(K.geq(n, 'gain', band), () => this.state.getGeq(n).gains?.get(Number(band)), address) if (cur === undefined) { this.log('warn', `${address}: current gain unknown (no reply from the console), not adjusted`) return false } return this.setGeqBand(n, band, cur + Number(delta)) } /** The OSC address of a channel parameter, or undefined (with a warning) when this entity kind does not have it. */ paramAddress(ent, n, param) { const address = addr.param(ent, n, this.sideOf(ent, n), param) if (!address) this.log('warn', `${ent} has no "${param}" parameter`) return address } setParamSwitch(ent, n, param, mode) { const address = this.paramAddress(ent, n, param) if (!address) return false return this.writeFlag(address, K.param(ent, n, param, this.sideOf(ent, n)), () => this.state.getParam(ent, n, param), mode) } setParamValue(ent, n, param, value) { const address = this.paramAddress(ent, n, param) if (!address || !this.addressOk(address)) return false const p = PARAMS[param] const v = clamp(Number(value), p.min, p.max) if (!Number.isFinite(v)) return false return this.write(address, [v], v) } async adjustParamValue(ent, n, param, delta) { const address = this.paramAddress(ent, n, param) if (!address || !this.addressOk(address)) return false const p = PARAMS[param] const cur = await this.current(K.param(ent, n, param, this.sideOf(ent, n)), () => this.state.getParam(ent, n, param), address) if (cur === undefined) { this.log('warn', `${address}: current value unknown (no reply from the console), not adjusted`) return false } const next = p.adjust === 'percent' ? Math.max(cur, p.min) * (1 + Number(delta) / 100) : cur + Number(delta) return this.setParamValue(ent, n, param, next) } fireSnapshot(index) { if (!Number.isInteger(index) || index < 0 || index > 9999) { this.log('warn', `Snapshot: "${index}" is not a snapshot index (a whole number from 0 to 9999)`) return false } const ok = this.sendOsc(addr.fireSnapshot, [{ i: index }]) this.later(() => this.enqueue([query(addr.currentSnapshot)]), 400) return ok } fireNextSnapshot() { const ok = this.sendOsc(addr.fireNext, [{ i: 0 }]) this.later(() => this.enqueue([query(addr.currentSnapshot)]), 400) return ok } firePrevSnapshot() { const ok = this.sendOsc(addr.firePrev, [{ i: 0 }]) this.later(() => this.enqueue([query(addr.currentSnapshot)]), 400) return ok } reload() { this.startLoad('reload') } forgetRelayClients() { this.relayClients.clear() this.publishRelay() } sendRaw(options) { if (!this.link) return false const address = String(options.address || '').trim() if (!address.startsWith('/')) { this.log('warn', `raw OSC: address must start with "/": ${address}`) return false } const v = String(options.value ?? '') let args = [] switch (options.argType) { case 'int': args = [{ i: Math.trunc(Number(v)) || 0 }] break case 'float': args = [Number(v) || 0] break case 'string': args = [v] break case 'bool': args = [/^(1|true|on|yes)$/i.test(v.trim())] break default: args = [] } this.log('info', `raw OSC -> ${address} ${JSON.stringify(args)}`) return this.link.send(address, args) } // ---------- loading ---------- /** Names first, then values for the entities that answered; one "load complete" line at the end. */ startLoad(reason) { if (!this.link) return const token = ++this.loadToken this.loading = true this.requested.clear() this.log('info', `Loading from the console (${reason})`) this.enqueue([...this.state.headerQueries(), ...this.state.nameQueries()]) this.whenIdle(() => this.later(() => { if (token !== this.loadToken) return const missing = this.state.nameQueries(true) if (!missing.length) return this.loadValues(token) // the console can drop queries; ask for the missing names once more this.enqueue(missing) this.whenIdle(() => this.later(() => this.loadValues(token), this.timing.settleMs)) }, this.timing.settleMs), ) } loadValues(token) { if (token !== this.loadToken) return this.enqueue(this.state.valueQueries()) this.whenIdle(() => this.later(() => this.finishLoad(token), this.timing.settleMs)) } finishLoad(token) { if (token !== this.loadToken) return const first = !this.state.loaded this.loading = false this.state.loaded = true const s = this.state.summary() this.log( 'info', `load complete: ${s.inputs} inputs, ${s.busses} busses (${s.aux} aux and ${s.group} group named), ${s.matrix} matrix outputs, ${s.matrixIn} matrix inputs, ` + `${s.cgs} control groups, ${s.geqs} graphic EQs, ${s.snapshots} snapshots`, ) if (!this.connectedFlag || s.total === 0) this.reloadOnReconnect = true if (first || this.rebuildPending) this.rebuildNow() else this.checkAllFeedbacks() // lazily loaded pairs asked for while the link was down are requested again this.resubscribeMeters() } // ---------- inbound ---------- onConsoleMessage(msg, rinfo) { if (msg.bundle) { for (const el of msg.elements) this.onConsoleMessage(el, rinfo) return } if (!this.connectedFlag) this.setConnected(true) this.learn('console', msg, rinfo) if (!msg.address) return const isMeter = msg.address === addr.meterValues if (this.config.debugLog && !isMeter) this.log('debug', `console -> ${describe(msg)}`) if (!isMeter && this.isStale(msg)) return const ev = this.state.applyMessage(msg.address, msg.args) if (ev) this.afterEvent(ev) } /** Re-check the feedback instances watching `key`; types in `fallbackTypes` only when none is known to. */ touch(key, fallbackTypes = []) { const ids = key ? this.watchers.get(key) : undefined if (ids && ids.size) for (const id of ids) this.pendingIds.add(id) else for (const type of fallbackTypes) this.pendingFeedbacks.add(type) this.scheduleFlush() } /** Register a feedback instance as watching `key` (replacing what it watched before, when its options changed). */ watch(fb, key) { const id = fb?.id if (!id || !key) return const prev = this.watching.get(id) if (prev === key) return if (prev) this.dropWatcher(prev, id) this.watching.set(id, key) let set = this.watchers.get(key) if (!set) { set = new Set() this.watchers.set(key, set) } set.add(id) } unwatch(fb) { const id = fb?.id const prev = id && this.watching.get(id) if (!prev) return this.dropWatcher(prev, id) this.watching.delete(id) } dropWatcher(key, id) { const set = this.watchers.get(key) if (!set) return set.delete(id) if (!set.size) this.watchers.delete(key) } /** Everything that follows a state change, from the console or from our own optimistic write. */ afterEvent(ev) { if (ev.kind === 'meters') { if (!ev.changed) return const now = Date.now() for (const [id, value] of ev.pairs) { if (!this.watchers.has(watchKey.meter(id))) continue this.meterDirtyIds.add(id) // within 20 dB of full scale on a level meter: a peak hold may be running (gain reduction meters have none) if (value <= 20 && this.meterWanted.get(id)?.meter === 'level') this.peakUntil.set(id, now + this.timing.peakWatchMs) } return } if (ev.key) this.notifyWaiters(ev.key) if (!ev.changed && !ev.structural) return this.queueVariables(getEventVariableValues(this, ev)) const change = ev.changed && this.state.loaded ? describeChange(this, ev) : '' if (change) this.queueVariables({ last_change: change }) this.touch(watchKeyForEvent(ev), feedbackIdsForEvent(ev)) if (ev.kind === 'snapCurrent' && this.state.loaded && !this.state.snapshots.list.has(ev.value)) this.enqueue([query(addr.snapshotNames)]) if (ev.structural) this.onStructural(ev) } onStructural(ev) { if (ev.kind === 'bussTypes' && this.state.loaded) { for (const n of this.state.existing('buss')) this.enqueue(this.state.bussValueQueries(n)) this.resubscribeMeters() // a buss meter lives on the buss's aux or group side } this.scheduleRebuild() } // ---------- definitions, variables, feedbacks ---------- scheduleRebuild() { if (this.loading || !this.state.loaded) { this.rebuildPending = true return } if (this.rebuildTimer) clearTimeout(this.rebuildTimer) this.rebuildTimer = setTimeout(() => { this.rebuildTimer = null this.rebuildNow() }, this.timing.rebuildDebounceMs) } rebuildNow() { if (this.rebuildTimer) clearTimeout(this.rebuildTimer) this.rebuildTimer = null this.rebuildPending = false this.rebuildDefinitions() } /** Push one kind of definition unless it is identical (callbacks aside) to what was pushed last. */ pushIfChanged(kind, value, set) { const json = JSON.stringify(value) if (this.lastDefs[kind] === json) return false this.lastDefs[kind] = json set() return true } rebuildDefinitions() { const actions = getActionDefinitions(this) const feedbacks = getFeedbackDefinitions(this) const vars = getVariableDefinitions(this) const { structure, presets, stats } = getPresetDefinitions(this) const changedActions = this.pushIfChanged('actions', actions, () => this.setActionDefinitions(actions)) const changedFeedbacks = this.pushIfChanged('feedbacks', feedbacks, () => this.setFeedbackDefinitions(feedbacks)) this.pushIfChanged('variables', vars, () => this.setVariableDefinitions(vars)) this.definedVariables = new Set(Object.keys(vars)) this.presetStats = stats const changedPresets = this.pushIfChanged('presets', { structure, presets }, () => this.setPresetDefinitions(structure, presets)) if (changedPresets && stats.dropped.length) { this.log('warn', `Too many presets (limit reached): left out ${stats.dropped.join(' and ')} (${stats.count} presets, ${stats.bytes} bytes)`) } this.queueVariables(getAllVariableValues(this)) this.publishRelay() this.flush() if (changedActions || changedFeedbacks) this.checkAllFeedbacks() } queueVariables(values) { Object.assign(this.pendingVars, values) this.scheduleFlush() } scheduleFlush() { if (this.flushTimer) return this.flushTimer = setTimeout(() => this.flush(), this.timing.flushMs) } flush() { if (this.flushTimer) clearTimeout(this.flushTimer) this.flushTimer = null const vars = {} for (const [id, v] of Object.entries(this.pendingVars)) if (this.definedVariables.has(id)) vars[id] = v this.pendingVars = {} const ids = [...this.pendingIds] this.pendingIds.clear() const types = [...this.pendingFeedbacks] this.pendingFeedbacks.clear() if (Object.keys(vars).length) this.setVariableValues(vars) if (ids.length) this.checkFeedbacksById(...ids) if (types.length) this.checkFeedbacks(...types) } publishRelay() { const live = this.relay ? this.relay.liveClients() : [] this.queueVariables({ relay_clients: live.length, relay_client_ips: live.map((c) => c.ip).join(', ') }) this.touch(watchKey.relay, ['relay_clients']) } // ---------- meters ---------- /** The [address, args] that subscribes one wanted meter; a buss meter follows the buss's current aux / group side. */ meterSubscriptionFor({ ent, n, meter }) { if (ent === 'input') return meterSubscription(n, meter) return outputMeterSubscription(ent, n, this.sideOf(ent, n)) } /** Subscribe to an input meter ('level', 'gate' or 'comp' gain reduction) the first time a meter feedback for it runs. */ subscribeInputMeter(n, meter = 'level') { this.subscribeMeter(meterId(n, meter), { ent: 'input', n, meter }) } /** Subscribe to a buss ('buss') or matrix output ('matrix') level meter, lazily like the input meters. */ subscribeOutputMeter(ent, n) { this.subscribeMeter(outputMeterId(ent, n), { ent, n, meter: 'level' }) } subscribeMeter(id, entry) { if (this.meterWanted.has(id)) return const sub = this.meterSubscriptionFor(entry) if (!this.addressOk(sub[0])) return this.meterWanted.set(id, entry) this.enqueue([sub]) this.ensureMeterTimer() } resubscribeMeters() { if (!this.meterWanted.size) return this.enqueue([...this.meterWanted.values()].map((entry) => this.meterSubscriptionFor(entry))) } scheduleMeterResubscribe() { if (!this.meterWanted.size) return if (this.resubscribeTimer) { clearTimeout(this.resubscribeTimer) this.timers.delete(this.resubscribeTimer) } this.resubscribeTimer = this.later(() => { this.resubscribeTimer = null this.resubscribeMeters() }, this.timing.resubscribeMs) } /** * Meter feedbacks are re-checked from this timer only (at most 10 Hz), and only the instances watching * a meter id that changed (or whose peak hold is still running) are asked to re-evaluate. */ ensureMeterTimer() { if (this.meterTimer) return this.meterTimer = setInterval(() => { const now = Date.now() for (const [id, until] of this.peakUntil) { this.meterDirtyIds.add(id) if (now >= until) this.peakUntil.delete(id) } if (!this.meterDirtyIds.size) return const ids = new Set() for (const id of this.meterDirtyIds) for (const fid of this.watchers.get(watchKey.meter(id)) || []) ids.add(fid) this.meterDirtyIds.clear() if (ids.size) this.checkFeedbacksById(...ids) }, this.timing.meterThrottleMs) } // ---------- debug learn log ---------- learn(source, msg, rinfo) { if (!this.config.debugLog) return if (!msg || !msg.address) { const key = `${source}:${msg?.bundle ? '#bundle' : 'raw'}` if (this.seen.has(key) || this.seen.size > LEARN_CAP) return this.seen.add(key) this.log('info', `learn ${source}: ${describe(msg)} from ${rinfo?.address}:${rinfo?.port}`) return } const key = `${source}:${foldPattern(msg.address)} ${msg.tags || ''}${msg.unpadded ? ' (unpadded)' : ''}` if (this.seen.has(key) || this.seen.size > LEARN_CAP) return this.seen.add(key) this.log('info', `learn ${source}: ${key} e.g. ${describe(msg)}`) } } export const UpgradeScripts = [upgradeV1Config]