// Live, read-only connections to your printers. // // One connection per printer stays open, so status arrives as the printer sends // it (every second or two while printing) instead of on a timer. The printer is // only ever asked for a full report — never sent anything that changes what it's // doing. Access codes are held in memory for the connection and never leave // this module. const mqtt = require('mqtt'); const { describeAms, describeTray, friendlyError, readStatus, PORT } = require('./printer'); const RECONNECT_MS = 15 * 1000; const FULL_REPORT_MS = 10 * 60 * 1000; // a fresh full report now and then, in case a change was missed const TRAYS_SETTLE_MS = 5 * 1000; // let a spool swap finish reporting before syncing it const connections = new Map(); // printer id → connection let onTrays = async () => null; let connect = (url, options) => mqtt.connect(url, options); // replaceable in tests // What's loaded, reduced to what decides whether spools need syncing. const traysKey = (trays) => JSON.stringify((trays || []) .map((t) => [t.unit, t.slot, t.empty, t.material, t.colour, t.uuid])); function allTrays(c) { return [...(c.ams || []), ...(c.external ? [c.external] : [])]; } // Which tray is feeding the toolhead. The printer sends `tray_now` as a number: // 255 for none, 254 for the external spool, otherwise unit × 4 + slot — where // "unit" is the AMS's id on most printers, or its position among the units. function activeTray(c) { const n = c.trayNow; if (n == null || Number.isNaN(n) || n === 255) return null; if (n === 254) return { external: true }; const ids = c.unitIds || []; const byId = Math.floor(n / 4); const unit = ids.includes(byId) ? byId : ids[byId]; return unit === undefined ? null : { unit, slot: n % 4 }; } function requestFullReport(c) { if (!c.client.connected) return; c.client.publish(`device/${c.serial}/request`, JSON.stringify({ pushing: { sequence_id: String(Date.now()), command: 'pushall' } })); } // Syncs spools with what's loaded now, and hands back what that did. async function flushTrays(c) { clearTimeout(c.traysTimer); c.traysTimer = null; c.pendingKey = null; const trays = allTrays(c); if (!c.ams && !c.external) return null; c.syncedKey = traysKey(trays); try { c.lastSync = await onTrays(c.id, trays); return c.lastSync; } catch (err) { console.warn(`Printer ${c.id} spool sync failed:`, err.message); return null; } } function handleMessage(c, payload) { let msg; try { msg = JSON.parse(payload.toString()); } catch { return; } const p = msg && msg.print; if (!p) return; c.reports += 1; c.updatedAt = new Date(); Object.assign(c.status, readStatus(p)); // AMS details come whole when they change; a report without them changes nothing. const units = p.ams && p.ams.ams; if (Array.isArray(units) && units.every((u) => Array.isArray(u.tray))) { c.ams = describeAms(units); c.unitIds = units.map((u) => Number(u.id)); } if (p.ams && p.ams.tray_now !== undefined) c.trayNow = Number(p.ams.tray_now); if (p.vt_tray) c.external = describeTray(p.vt_tray, null); // A printer mid-print reports every second or so, so only a change in what's // loaded restarts the wait — otherwise the sync would never come round. const key = (c.ams || c.external) ? traysKey(allTrays(c)) : null; if (key && key !== c.syncedKey && key !== c.pendingKey) { c.pendingKey = key; clearTimeout(c.traysTimer); c.traysTimer = setTimeout(() => flushTrays(c), TRAYS_SETTLE_MS); } // Someone waiting for a full report is answered by one with the trays in it. if (Array.isArray(units) || p.vt_tray) for (const resolve of c.waiters.splice(0)) resolve(); } function open(p) { const c = { id: p.id, serial: p.serial, key: [p.host, p.serial, p.access_code].join('\n'), connected: false, error: null, reports: 0, updatedAt: null, connectedAt: null, status: {}, ams: null, external: null, syncedKey: null, pendingKey: null, trayNow: null, unitIds: [], lastSync: null, traysTimer: null, waiters: [], }; c.client = connect(`mqtts://${p.host}:${PORT}`, { username: 'bblp', password: p.access_code, rejectUnauthorized: false, // the printer's certificate is self-signed connectTimeout: 10 * 1000, reconnectPeriod: RECONNECT_MS, keepalive: 30, clientId: `printhive-${p.id}-${Math.random().toString(16).slice(2, 8)}`, }); c.client.on('connect', () => { c.connected = true; c.error = null; c.connectedAt = new Date(); c.client.subscribe(`device/${c.serial}/report`, (err) => { if (err) { c.error = `The printer wouldn’t let Printhive listen: ${err.message}`; return; } requestFullReport(c); }); }); c.client.on('message', (_topic, payload) => handleMessage(c, payload)); c.client.on('error', (err) => { c.error = friendlyError(err); }); c.client.on('close', () => { c.connected = false; }); c.client.on('offline', () => { c.connected = false; }); c.fullReportTimer = setInterval(() => requestFullReport(c), FULL_REPORT_MS); c.fullReportTimer.unref(); return c; } function close(c) { clearInterval(c.fullReportTimer); clearTimeout(c.traysTimer); for (const resolve of c.waiters.splice(0)) resolve(); c.client.end(true); } // Brings the connections in line with the printers you have: new ones opened, // removed ones closed, and any whose address, serial or code changed reopened. function reconcile(printerRows) { const wanted = new Map(); for (const p of printerRows) { if (p.host && p.serial && p.access_code) wanted.set(p.id, p); } for (const [id, c] of connections) { const p = wanted.get(id); if (!p || [p.host, p.serial, p.access_code].join('\n') !== c.key) { close(c); connections.delete(id); } } for (const [id, p] of wanted) { if (!connections.has(id)) connections.set(id, open(p)); } } // What one printer is doing, as last reported. Nothing here identifies a spool // or carries a credential. function snapshot(id) { const c = connections.get(id); if (!c) return { live: false }; const loaded = allTrays(c).map((t) => ({ unit: t.unit, slot: t.slot, empty: t.empty, material: t.material, colour: t.colour, tagged: Boolean(t.uuid), })); return { live: true, connected: c.connected, error: c.connected ? null : c.error, updatedAt: c.updatedAt, status: { ...c.status, active: activeTray(c) }, loaded: c.ams || c.external ? loaded : null, }; } function snapshots() { const out = {}; for (const id of connections.keys()) out[id] = snapshot(id); return out; } // Asks for a full report now, waits for it, and syncs spools from it — what // "Test & sync" does. Uses the open connection rather than making another. async function refresh(id, waitMs = 8000) { const c = connections.get(id); if (!c) return null; const started = Date.now(); const before = c.reports; if (!c.connected && !c.client.reconnecting) c.client.reconnect(); await new Promise((resolve) => { const timer = setTimeout(resolve, waitMs); const ask = () => { if (c.connected) { requestFullReport(c); return; } if (Date.now() - started < waitMs) setTimeout(ask, 250); }; c.waiters.push(() => { clearTimeout(timer); setTimeout(resolve, 1000); }); // let the rest of the report arrive ask(); }); const sync = c.reports > before ? await flushTrays(c) : null; return { connected: c.connected, reports: c.reports - before, printer: { state: c.status.state, progress: c.status.progress }, ams: c.ams, external: c.external, error: c.connected ? null : (c.error || 'Couldn’t reach the printer.'), sync, ms: Date.now() - started, }; } function setTraysHandler(fn) { onTrays = fn; } function stopAll() { for (const c of connections.values()) close(c); connections.clear(); } // For tests: connect with a stand-in instead of the network. function setConnector(fn) { connect = fn; } module.exports = { reconcile, snapshot, snapshots, refresh, setTraysHandler, stopAll, setConnector };