Phase 1 wave 2: BaseDomain + commandRegistry + statusUpdater
- src/domain/BaseDomain.js — base class for every specificClass; wires emitter/config/logger/measurements/childRouter - src/nodered/commandRegistry.js — declarative msg.topic dispatch with alias deprecation - src/nodered/statusUpdater.js — 1Hz status badge poller with error-resilient loop Additive. 43 new tests; all 99 basic tests still green. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
znetsixe
156
src/nodered/commandRegistry.js
Normal file
156
src/nodered/commandRegistry.js
Normal file
@@ -0,0 +1,156 @@
|
||||
'use strict';
|
||||
|
||||
// Declarative dispatch for a node's input topics. Each node declares its
|
||||
// commands as an array of descriptors; the registry builds an O(1) lookup
|
||||
// keyed by canonical topic + alias, validates the payload against a small
|
||||
// shape schema, and invokes the handler. Replaces the per-node ~100-line
|
||||
// `switch (msg.topic)` block in nodeClass._attachInputHandler.
|
||||
//
|
||||
// Lightweight on purpose: the schema is a typeof-check ladder, not full
|
||||
// JSON-Schema. Anything richer belongs in the handler itself, which has
|
||||
// access to logger via ctx.
|
||||
|
||||
const SCALAR_TYPES = new Set(['string', 'number', 'boolean', 'object', 'any']);
|
||||
|
||||
class CommandRegistry {
|
||||
constructor(commands, options = {}) {
|
||||
if (!Array.isArray(commands)) {
|
||||
throw new TypeError('CommandRegistry requires an array of command descriptors');
|
||||
}
|
||||
this._logger = options.logger || null;
|
||||
this._byKey = new Map(); // topic-or-alias -> descriptor
|
||||
this._canonicalByAlias = new Map();
|
||||
this._descriptors = [];
|
||||
this._deprecationCounts = new Map();
|
||||
this._deprecationLogged = new Set();
|
||||
for (const cmd of commands) this._register(cmd);
|
||||
}
|
||||
|
||||
_register(cmd) {
|
||||
if (!cmd || typeof cmd.topic !== 'string' || cmd.topic.length === 0) {
|
||||
throw new TypeError('command descriptor requires a non-empty string topic');
|
||||
}
|
||||
if (typeof cmd.handler !== 'function') {
|
||||
throw new TypeError(`command '${cmd.topic}' requires a handler function`);
|
||||
}
|
||||
if (this._byKey.has(cmd.topic)) {
|
||||
throw new Error(`duplicate command topic '${cmd.topic}'`);
|
||||
}
|
||||
const aliases = Array.isArray(cmd.aliases) ? cmd.aliases.slice() : [];
|
||||
for (const alias of aliases) {
|
||||
if (typeof alias !== 'string' || alias.length === 0) {
|
||||
throw new TypeError(`command '${cmd.topic}' has an invalid alias`);
|
||||
}
|
||||
if (this._byKey.has(alias)) {
|
||||
throw new Error(`alias '${alias}' for '${cmd.topic}' collides with existing topic or alias`);
|
||||
}
|
||||
}
|
||||
const descriptor = {
|
||||
topic: cmd.topic,
|
||||
aliases,
|
||||
payloadSchema: cmd.payloadSchema || null,
|
||||
handler: cmd.handler,
|
||||
};
|
||||
this._byKey.set(cmd.topic, descriptor);
|
||||
for (const alias of aliases) {
|
||||
this._byKey.set(alias, descriptor);
|
||||
this._canonicalByAlias.set(alias, cmd.topic);
|
||||
}
|
||||
this._descriptors.push(descriptor);
|
||||
}
|
||||
|
||||
has(topic) {
|
||||
return typeof topic === 'string' && this._byKey.has(topic);
|
||||
}
|
||||
|
||||
canonical(topic) {
|
||||
if (typeof topic !== 'string') return topic;
|
||||
return this._canonicalByAlias.get(topic) || topic;
|
||||
}
|
||||
|
||||
list() {
|
||||
// Strip handler so callers can safely log / serialise the result
|
||||
// (handler functions are noisy and not contract-relevant).
|
||||
return this._descriptors.map((d) => ({
|
||||
topic: d.topic,
|
||||
aliases: d.aliases.slice(),
|
||||
payloadSchema: d.payloadSchema,
|
||||
}));
|
||||
}
|
||||
|
||||
deprecationStats() {
|
||||
const out = {};
|
||||
for (const [alias, count] of this._deprecationCounts) out[alias] = count;
|
||||
return out;
|
||||
}
|
||||
|
||||
async dispatch(msg, source, ctx) {
|
||||
const log = this._loggerFor(ctx);
|
||||
const topic = msg && typeof msg.topic === 'string' ? msg.topic : null;
|
||||
if (!topic) {
|
||||
log.warn?.('commandRegistry: msg has no topic; ignoring');
|
||||
return;
|
||||
}
|
||||
const descriptor = this._byKey.get(topic);
|
||||
if (!descriptor) {
|
||||
log.warn?.(`commandRegistry: unknown topic '${topic}'`);
|
||||
return;
|
||||
}
|
||||
if (topic !== descriptor.topic) this._noteAlias(topic, descriptor.topic, log);
|
||||
if (!this._validatePayload(descriptor, msg, log)) return;
|
||||
return descriptor.handler(source, msg, ctx);
|
||||
}
|
||||
|
||||
_noteAlias(alias, canonical, log) {
|
||||
const prev = this._deprecationCounts.get(alias) || 0;
|
||||
this._deprecationCounts.set(alias, prev + 1);
|
||||
if (this._deprecationLogged.has(alias)) return;
|
||||
this._deprecationLogged.add(alias);
|
||||
log.warn?.(`topic '${alias}' is deprecated; use '${canonical}'`);
|
||||
}
|
||||
|
||||
_validatePayload(descriptor, msg, log) {
|
||||
const schema = descriptor.payloadSchema;
|
||||
if (!schema) return true;
|
||||
const payload = msg.payload;
|
||||
const type = schema.type || 'any';
|
||||
if (!SCALAR_TYPES.has(type)) {
|
||||
log.warn?.(`commandRegistry: command '${descriptor.topic}' has unknown schema type '${type}'`);
|
||||
return true;
|
||||
}
|
||||
if (type === 'any') return true;
|
||||
// typeof null === 'object' — explicit null fails an object schema.
|
||||
if (type === 'object') {
|
||||
if (payload === null || typeof payload !== 'object') {
|
||||
log.warn?.(`commandRegistry: '${descriptor.topic}' expected object payload, got ${payload === null ? 'null' : typeof payload}`);
|
||||
return false;
|
||||
}
|
||||
} else if (typeof payload !== type) {
|
||||
log.warn?.(`commandRegistry: '${descriptor.topic}' expected ${type} payload, got ${typeof payload}`);
|
||||
return false;
|
||||
}
|
||||
if (type === 'object' && schema.properties && typeof schema.properties === 'object') {
|
||||
for (const [key, expected] of Object.entries(schema.properties)) {
|
||||
if (!(key in payload)) continue; // missing keys allowed
|
||||
if (typeof payload[key] !== expected) {
|
||||
log.warn?.(`commandRegistry: '${descriptor.topic}' payload.${key} expected ${expected}, got ${typeof payload[key]}`);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
_loggerFor(ctx) {
|
||||
const candidate = (ctx && ctx.logger) || this._logger;
|
||||
return candidate || NOOP_LOGGER;
|
||||
}
|
||||
}
|
||||
|
||||
const NOOP_LOGGER = { warn() {}, error() {}, info() {}, debug() {} };
|
||||
|
||||
function createRegistry(commands, options) {
|
||||
return new CommandRegistry(commands, options);
|
||||
}
|
||||
|
||||
module.exports = { createRegistry, CommandRegistry };
|
||||
90
src/nodered/statusUpdater.js
Normal file
90
src/nodered/statusUpdater.js
Normal file
@@ -0,0 +1,90 @@
|
||||
/**
|
||||
* StatusUpdater — periodic Node-RED status badge poller.
|
||||
*
|
||||
* Replaces the per-node `_statusInterval` boilerplate (e.g. pumpingStation
|
||||
* nodeClass lines 160-171) with one class. The adapter constructs it once
|
||||
* with a `node` (Node-RED handle) and a `source` (the domain), and the
|
||||
* loop drives `node.status(source.getStatusBadge())` at a fixed cadence.
|
||||
*
|
||||
* Errors thrown from the domain become a red error badge instead of
|
||||
* crashing the interval — operators see the failure in the editor.
|
||||
*
|
||||
* See CONTRACTS.md §7 for the badge shape; statusBadge.js for the helpers.
|
||||
*/
|
||||
|
||||
'use strict';
|
||||
|
||||
const { statusBadge } = require('./statusBadge');
|
||||
|
||||
const CLEAR_BADGE = {};
|
||||
|
||||
class StatusUpdater {
|
||||
constructor({ node, source, intervalMs, logger } = {}) {
|
||||
if (!node || typeof node.status !== 'function') {
|
||||
throw new Error('StatusUpdater: node must expose a .status(badge) method');
|
||||
}
|
||||
if (!source || typeof source.getStatusBadge !== 'function') {
|
||||
throw new Error('StatusUpdater: source must expose a .getStatusBadge() method');
|
||||
}
|
||||
this._node = node;
|
||||
this._source = source;
|
||||
this._intervalMs = Number.isFinite(intervalMs) ? intervalMs : 0;
|
||||
this._logger = logger || null;
|
||||
this._timer = null;
|
||||
}
|
||||
|
||||
get isRunning() {
|
||||
return this._timer !== null;
|
||||
}
|
||||
|
||||
start() {
|
||||
// intervalMs=0 keeps unit tests / headless harnesses silent.
|
||||
if (this._intervalMs <= 0) return;
|
||||
if (this._timer !== null) return;
|
||||
this._timer = setInterval(() => this._tick(), this._intervalMs);
|
||||
}
|
||||
|
||||
stop() {
|
||||
if (this._timer !== null) {
|
||||
clearInterval(this._timer);
|
||||
this._timer = null;
|
||||
}
|
||||
// Wipe the badge so a stale label doesn't linger in the editor
|
||||
// after the node is closed/redeployed.
|
||||
try { this._node.status(CLEAR_BADGE); } catch (_) { /* best effort */ }
|
||||
}
|
||||
|
||||
_tick() {
|
||||
let badge;
|
||||
try {
|
||||
badge = this._source.getStatusBadge();
|
||||
} catch (err) {
|
||||
const msg = err && err.message ? err.message : String(err);
|
||||
if (this._logger && typeof this._logger.error === 'function') {
|
||||
this._logger.error(`StatusUpdater: getStatusBadge threw: ${msg}`);
|
||||
}
|
||||
this._safeApply(statusBadge.error(msg));
|
||||
return;
|
||||
}
|
||||
if (badge == null) {
|
||||
this._safeApply(CLEAR_BADGE);
|
||||
return;
|
||||
}
|
||||
this._safeApply(badge);
|
||||
}
|
||||
|
||||
_safeApply(badge) {
|
||||
try {
|
||||
this._node.status(badge);
|
||||
} catch (err) {
|
||||
// node.status itself failing is exotic (e.g. node already
|
||||
// closed). Log once per tick; the next tick will retry.
|
||||
if (this._logger && typeof this._logger.error === 'function') {
|
||||
const msg = err && err.message ? err.message : String(err);
|
||||
this._logger.error(`StatusUpdater: node.status threw: ${msg}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
module.exports = { StatusUpdater };
|
||||
Reference in New Issue
Block a user