// One light unit's connection. A port of the lessons in
// blender-addon/cthuloid_lights/net.py, where each was paid for:
//
// * Each unit owns its own socket and timers, so an unreachable unit only
// ever delays itself. (The addon once served every fixture from one thread,
// and three unplugged units' ~5 s .local lookups starved the live one.)
// * Latest state only. `set()` replaces the pending look; if the browser
// outruns the wire, intermediate frames are dropped, never queued, so a
// slider drag can't build a backlog that plays back late.
// * Rate limited (the firmware is single-threaded) and deduplicated.
// * Reconnect with backoff; on reconnect the desired look is re-asserted,
// because the unit may have rebooted and cannot be assumed to hold it.
// * Keepalive with `hostname`, the one command that touches no pin state, so
// the firmware's 5-minute idle drop never fires on a held look.
//
// Nothing here connects until open() (the Connect switch) or sendOnce()
// (Blackout while disconnected).
//
// Proto 3 (firmware with shared control) adds, per the design (§5):
// * per-KEY latest-only sending: 'look' (whole-unit pwm/pwm2) and one key per
// slot ('slot0'..'slot3'), so a drag on one slot never drops another's frame;
// * grant() matched to its reply -- the firmware answers commands in order;
// * release() flushes that slot's pending frame first, or the final value
// would reach the unit after the release and be refused as not held;
// * held / free / values broadcasts as events. After a reconnect every grant
// is gone (the unit freed them on disconnect): 'reset' is emitted and slot
// frames are never re-asserted, only the 'look'.
import EventEmitter from 'bare-events'
import net from 'bare-net'
import * as P from './protocol.js'
const GRANT_TIMEOUT_MS = 3000
const KEEPALIVE_PAYLOAD = 'hostname\n'
export class Unit extends EventEmitter {
constructor ({
id,
host,
port = 1337,
minIntervalMs = 40,
keepaliveMs = 60000,
connectTimeoutMs = 3000,
backoffStartMs = 1000,
backoffMaxMs = 15000
}) {
super()
this.id = id
this.host = host
this.port = port
this.minIntervalMs = minIntervalMs
this.keepaliveMs = keepaliveMs
this.connectTimeoutMs = connectTimeoutMs
this.backoffStartMs = backoffStartMs
this.backoffMaxMs = backoffMaxMs
this.status = 'off' // off | connecting | connected | unreachable
this.error = ''
this.greetingName = null
this.actual = null // last readback: `state` reply, kept current by `values`
this.proto = 0 // from `hello`; 0 until seen (proto <= 2 never sends one)
this.epoch = 0
this.pk = ''
this._grants = [] // pending grant() calls, FIFO: { need, got: [], resolve, timer }
this._wanted = false // open() called and not close()d
this._once = false // sendOnce in progress: deliver, then let go
this._sock = null
this._connected = false
this._buf = ''
this._pending = new Map() // key -> payload not yet written
this._lastSent = new Map() // key -> payload last written
this._force = false
this._lastTx = 0
this._backoff = backoffStartMs
this._retryTimer = null
this._flushTimer = null
this._connectTimer = null
this._keepaliveTimer = null
}
// ---- control -----------------------------------------------------------
open () {
this._once = false
if (this._wanted) return
this._wanted = true
this._backoff = this.backoffStartMs
if (!this._sock) this._connect()
}
close () {
this._wanted = false
this._once = false
this._teardown()
clearTimeout(this._retryTimer)
this._retryTimer = null
this._lastSent.clear()
this._pending.clear()
this._setStatus('off', '')
}
// Latest payload for a key ('look' = the whole unit). Never blocks; delivered
// when connected, latest-only per key, identical payloads skipped.
set (payload, { force = false, key = 'look' } = {}) {
if (!payload) return
if (!force && this._lastSent.get(key) === payload) {
this._pending.delete(key)
return
}
this._pending.set(key, payload)
if (force) this._force = true
this._flush()
}
// ---- proto 3 ------------------------------------------------------------
// Ask the unit for a slot (0..3) or 'all'. Resolves { granted: [{slot, epoch,
// n}, ...] } or { busy: { slot, holder } } or { error }.
grant (slot, holder) {
return new Promise((resolve) => {
if (!this._connected || this.proto < 3) return resolve({ error: 'unit not connected with proto 3' })
const req = { need: slot === 'all' ? 4 : 1, got: [], resolve, timer: null }
req.timer = setTimeout(() => this._settleGrant(req, { error: 'no reply to grant' }), GRANT_TIMEOUT_MS)
this._grants.push(req)
this.command(P.grantCommand(slot, holder))
})
}
// A drag frame for a held slot: latest-only, rate limited with everything else.
sendSlot (slot, grant, bgr, fadeMs) {
this.set(P.slotCommand(slot, grant.epoch, grant.n, bgr, fadeMs), { key: 'slot' + slot })
}
// End a grant. Any frame still pending for the slot goes out FIRST.
release (slot) {
if (!this._connected) return false
const slots = slot === 'all' ? [0, 1, 2, 3] : [slot]
let out = ''
for (const s of slots) {
const key = 'slot' + s
if (this._pending.has(key)) {
out += this._pending.get(key)
this._pending.delete(key)
}
this._lastSent.delete(key)
}
return this.command(out + P.releaseCommand(slot))
}
_settleGrant (req, result) {
const i = this._grants.indexOf(req)
if (i === -1) return
this._grants.splice(i, 1)
clearTimeout(req.timer)
req.resolve(result)
}
_failGrants (error) {
for (const req of this._grants.slice()) this._settleGrant(req, { error })
}
// One-shot delivery while not open (Blackout when disconnected): one
// connection attempt, deliver, close. No retry loop is left behind.
sendOnce (payload) {
if (this._wanted) return this.set(payload, { force: true })
this._once = true
this._pending.set('look', payload)
this._force = true
if (!this._sock) this._connect()
else this._flush()
}
// A one-off command beside the look (gamma, freq, state). Unlike set() it
// does not replace the pending look, and it is dropped if not connected:
// these are re-pushed on demand, not state that must survive a reconnect.
command (payload) {
if (!this._connected) return false
this._sock.write(payload)
this._lastTx = Date.now()
return true
}
requestState () {
return this.command('state\n')
}
// ---- connection --------------------------------------------------------
_connect () {
this._setStatus('connecting', '')
this._connected = false
this._buf = ''
const sock = net.createConnection({ port: this.port, host: this.host, noDelay: true })
this._sock = sock
this._connectTimer = setTimeout(() => {
if (this._sock === sock && !this._connected) sock.destroy(new Error('connect timeout'))
}, this.connectTimeoutMs)
sock.on('connect', () => {
if (this._sock !== sock) return
clearTimeout(this._connectTimer)
this._connected = true
this._backoff = this.backoffStartMs
this._lastTx = Date.now()
this._setStatus('connected', '')
// The unit may have rebooted or been dropped mid-look: re-assert the
// look. Slot frames are NOT re-asserted -- their grants died with the
// old socket, so the unit would refuse them.
const look = this._pending.get('look') || this._lastSent.get('look')
this._lastSent.clear()
this._pending.clear()
if (look) this._pending.set('look', look)
this._flush()
this._armKeepalive()
})
sock.on('data', (d) => this._onData(d))
sock.on('error', (err) => {
if (this._sock === sock) this.error = err.message || String(err)
})
sock.on('close', () => {
if (this._sock !== sock) return
const wasConnected = this._connected
this._teardown()
this._failGrants('connection closed')
if (wasConnected) this.emit('reset')
if (this._once) {
// One attempt only. A once-send that failed to connect is dropped.
this._once = false
this._setStatus('off', wasConnected ? '' : this.error)
return
}
if (!this._wanted) return
this._setStatus('unreachable', this.error || (wasConnected ? 'connection closed' : 'connect failed'))
for (const k of [...this._pending.keys()]) if (k !== 'look') this._pending.delete(k)
this._retryTimer = setTimeout(() => {
this._retryTimer = null
if (this._wanted && !this._sock) this._connect()
}, this._backoff)
this._backoff = Math.min(this._backoff * 2, this.backoffMaxMs)
})
}
_teardown () {
clearTimeout(this._connectTimer)
clearTimeout(this._flushTimer)
clearInterval(this._keepaliveTimer)
this._flushTimer = null
this._keepaliveTimer = null
const sock = this._sock
this._sock = null
this._connected = false
if (sock) sock.destroy()
}
_onData (d) {
this._buf += d.toString()
let i
while ((i = this._buf.indexOf('\n')) !== -1) {
const line = this._buf.slice(0, i).trim()
this._buf = this._buf.slice(i + 1)
if (!line) continue
this._onLine(line)
}
// A unit that never sends a newline must not grow this without bound.
if (this._buf.length > 4096) this._buf = ''
}
_onLine (line) {
let x
if ((x = P.parseGreeting(line)) !== null) {
this.greetingName = x
this.emit('greeting', x)
} else if ((x = P.parseHello(line)) !== null) {
this.proto = x.proto
this.epoch = x.epoch
this.pk = x.pk
this.emit('hello', x)
} else if ((x = P.parseState(line)) !== null) {
this.actual = x
this.emit('state', x)
} else if ((x = P.parseValues(line)) !== null) {
this.actual = { ...(this.actual || { fading: false, freq: 0, gamma: null }), values: x }
this.emit('values', x)
} else if ((x = P.parseGranted(line)) !== null) {
const req = this._grants[0]
if (!req) return
req.got.push(x)
if (req.got.length >= req.need) this._settleGrant(req, { granted: req.got })
} else if ((x = P.parseBusy(line)) !== null) {
if (this._grants[0]) this._settleGrant(this._grants[0], { busy: x })
} else if ((x = P.parseHeld(line)) !== null) {
this.emit('held', x)
} else if ((x = P.parseFree(line)) !== null) {
this.emit('free', x)
} else if (line.startsWith('ERR grant')) {
if (this._grants[0]) this._settleGrant(this._grants[0], { error: line })
} else if (line.startsWith('ERR notheld')) {
this.emit('notheld', line)
}
}
// ---- sending -----------------------------------------------------------
_flush () {
if (!this._connected || this._pending.size === 0 || this._flushTimer) return
const wait = this.minIntervalMs - (Date.now() - this._lastTx)
if (!this._force && wait > 0) {
this._flushTimer = setTimeout(() => {
this._flushTimer = null
this._flush()
}, wait)
return
}
// Everything pending goes in one write; each key's latest only.
let payload = ''
for (const [key, p] of this._pending) {
payload += p
this._lastSent.set(key, p)
}
this._pending.clear()
this._sock.write(payload)
this._lastTx = Date.now()
this._force = false
if (this._once) {
// Let the bytes leave before closing; end() flushes, then 'close' fires.
this._once = false
this._sock.end()
}
}
_armKeepalive () {
clearInterval(this._keepaliveTimer)
const every = Math.max(1000, Math.floor(this.keepaliveMs / 4))
this._keepaliveTimer = setInterval(() => {
if (this._connected && Date.now() - this._lastTx >= this.keepaliveMs) {
this._sock.write(KEEPALIVE_PAYLOAD)
this._lastTx = Date.now()
}
}, every)
}
_setStatus (status, error) {
if (this.status === status && this.error === error) return
this.status = status
this.error = error
this.emit('status', status, error)
}
}