'use strict'; // Serialises an async dispatch so that high-frequency callers cannot stack // up overlapping invocations. Intermediate values are dropped — only the // most recent fire()/fireAndWait() during an in-flight dispatch is replayed // afterwards. Extracted from machineGroupControl's _dispatchInFlight + // _delayedCall pattern so MGC, pumpingStation, valveGroupControl etc. can // share it. // // fire(value) — never blocks; returns void. // fireAndWait(value) — returns a promise that settles when THIS value's // dispatch runs to completion. If a later fireAndWait // arrives during the in-flight call and supersedes // this one in the pending slot, the returned promise // RESOLVES with { superseded: true } instead of // rejecting — callers can branch on a sentinel // without try/catch. The dispatch's own return value // (when not superseded) is forwarded as the resolution. const SUPERSEDED = Object.freeze({ superseded: true }); class LatestWinsGate { constructor(asyncDispatchFn, options = {}) { if (typeof asyncDispatchFn !== 'function') { throw new TypeError('LatestWinsGate requires an async dispatch function'); } this._dispatch = asyncDispatchFn; this._logger = options.logger || null; this._inFlight = false; this._pending = null; // { value, ctx, settle? } | null this._drainResolvers = []; // resolved when idle again this.lastError = null; } // 0 = idle, 1 = running with no pending, 2 = running with pending. get size() { if (!this._inFlight) return 0; return this._pending ? 2 : 1; } // Never blocks. If a dispatch is in flight, the latest value is parked; // older parked values are silently overwritten. fire(value, ctx) { if (this._inFlight) { this._supersedePending(); this._pending = { value, ctx, settle: null }; return; } this._run(value, ctx, null); } // Returns a promise that resolves when THIS fire's dispatch settles. // If this fire gets overwritten while parked, resolves with the // SUPERSEDED sentinel ({ superseded: true }) — callers branch on // result.superseded === true without try/catch. fireAndWait(value, ctx) { return new Promise((resolve) => { const settle = resolve; if (this._inFlight) { this._supersedePending(); this._pending = { value, ctx, settle }; return; } this._run(value, ctx, settle); }); } drain() { if (!this._inFlight && !this._pending) return Promise.resolve(); return new Promise((resolve) => { this._drainResolvers.push(resolve); }); } _supersedePending() { const prev = this._pending; if (prev && typeof prev.settle === 'function') prev.settle(SUPERSEDED); this._pending = null; } _run(value, ctx, settle) { this._inFlight = true; // Kick the dispatch on a microtask so fire()/fireAndWait() always // return synchronously, even if _dispatch resolves immediately. Promise.resolve() .then(() => this._dispatch(value, ctx)) .then((result) => { if (typeof settle === 'function') settle(result); }, (err) => { this.lastError = err; if (this._logger && typeof this._logger.error === 'function') { this._logger.error(err); } // Resolve (not reject) so fireAndWait callers don't need // try/catch. Dispatch errors stay observable via lastError. if (typeof settle === 'function') settle(undefined); }) .then(() => this._afterDispatch()); } _afterDispatch() { this._inFlight = false; if (this._pending) { const { value, ctx, settle } = this._pending; this._pending = null; this._run(value, ctx, settle); return; } // Idle — release any drain() waiters. const waiters = this._drainResolvers; this._drainResolvers = []; for (const r of waiters) r(); } } LatestWinsGate.SUPERSEDED = SUPERSEDED; module.exports = LatestWinsGate;