refactor(engine): unify commands and orchestrators
Replace the CommandHost with a handler registry keyed by message type. Message is a permissive base type the host extends; CommandResult is a discriminated union on the type suffix. Commands and orchestrators are the same async shape taking a RunContext with an AbortSignal; runCommand emits :done/:cancel/:error.
This commit is contained in:
@@ -1,40 +1,21 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
import { Engine, type CommandHost, type Message } from './engine.js';
|
||||
|
||||
/** A minimal command host that runs a `move` and emits `move:done`. */
|
||||
function moveHost(): CommandHost & { moves: Message[] } {
|
||||
const handlers = new Map<string, (msg: Message) => void>();
|
||||
const moves: Message[] = [];
|
||||
return {
|
||||
moves,
|
||||
on(type, handler) {
|
||||
handlers.set(type, handler);
|
||||
return () => handlers.delete(type);
|
||||
},
|
||||
dispatch(msg) {
|
||||
if (msg.type === 'move') {
|
||||
moves.push(msg);
|
||||
// The command's completion is itself a message.
|
||||
this.dispatch({ type: 'move:done', data: msg.data });
|
||||
} else if (msg.type === 'move:done') {
|
||||
handlers.get('move:done')?.(msg);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
import { Engine, type Message } from './engine.js';
|
||||
|
||||
describe('Engine', () => {
|
||||
it('dispatches a message to the command host on tick', () => {
|
||||
const host = moveHost();
|
||||
const engine = new Engine(host);
|
||||
it('dispatches a message to handlers registered for its type', () => {
|
||||
const engine = new Engine();
|
||||
const seen: Message[] = [];
|
||||
engine.on('move', (m) => seen.push(m));
|
||||
engine.enqueue({ type: 'move', data: { part: 'a', to: '/grid/5/5' } });
|
||||
engine.tick();
|
||||
expect(host.moves).toHaveLength(1);
|
||||
expect(seen).toHaveLength(1);
|
||||
expect(seen[0]!.data).toEqual({ part: 'a', to: '/grid/5/5' });
|
||||
});
|
||||
|
||||
it('a trigger reacts to a message and emits on the next tick', () => {
|
||||
const host = moveHost();
|
||||
const engine = new Engine(host);
|
||||
const engine = new Engine();
|
||||
const seen: Message[] = [];
|
||||
engine.on('move', (m) => seen.push(m));
|
||||
engine.registerTrigger({
|
||||
type: 'tap',
|
||||
id: 'draw',
|
||||
@@ -44,14 +25,48 @@ describe('Engine', () => {
|
||||
|
||||
engine.enqueue({ type: 'tap', data: { part: 'carcassonne:tile#a', trigger: 'draw' } });
|
||||
engine.tick(); // tap processed; move emitted to the next tick
|
||||
expect(host.moves).toHaveLength(0);
|
||||
expect(seen).toHaveLength(0);
|
||||
engine.tick(); // move runs
|
||||
expect(host.moves).toHaveLength(1);
|
||||
expect(seen).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('runCommand runs the command and emits its result message', async () => {
|
||||
const engine = new Engine();
|
||||
const results: Message[] = [];
|
||||
engine.on('move:done', (m) => results.push(m));
|
||||
engine.on('move:error', (m) => results.push(m));
|
||||
|
||||
engine.runCommand('move', async ({ args }) => {
|
||||
expect(args).toEqual({ part: 'a', to: '/grid/5/5' });
|
||||
return 'moved';
|
||||
});
|
||||
|
||||
engine.enqueue({ type: 'move', data: { part: 'a', to: '/grid/5/5' } });
|
||||
engine.tick(); // command runs; result emitted to the next tick
|
||||
await Promise.resolve(); // flush the command's async resolution
|
||||
engine.tick(); // result processed
|
||||
expect(results).toEqual([{ type: 'move:done', data: 'moved' }]);
|
||||
});
|
||||
|
||||
it('runCommand emits :error when the command throws', async () => {
|
||||
const engine = new Engine();
|
||||
const results: Message[] = [];
|
||||
engine.on('move:error', (m) => results.push(m));
|
||||
|
||||
engine.runCommand('move', async () => {
|
||||
throw new Error('bad path');
|
||||
});
|
||||
|
||||
engine.enqueue({ type: 'move', data: {} });
|
||||
engine.tick();
|
||||
await Promise.resolve();
|
||||
await Promise.resolve(); // rejection hops through .then before .catch
|
||||
engine.tick();
|
||||
expect(results).toEqual([{ type: 'move:error', error: new Error('bad path') }]);
|
||||
});
|
||||
|
||||
it('an orchestrator awaits a matching message and resumes on tick', async () => {
|
||||
const host = moveHost();
|
||||
const engine = new Engine(host);
|
||||
const engine = new Engine();
|
||||
const log: string[] = [];
|
||||
|
||||
const done = engine.runOrchestrator(async (ctx) => {
|
||||
@@ -71,6 +86,27 @@ describe('Engine', () => {
|
||||
expect(log).toEqual(['start', 'tap:carcassonne:tile#a']);
|
||||
// The move emitted by the orchestrator runs on the next tick.
|
||||
engine.tick();
|
||||
expect(host.moves).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('an orchestrator wait rejects when its signal aborts', async () => {
|
||||
const engine = new Engine();
|
||||
const controller = new AbortController();
|
||||
const log: string[] = [];
|
||||
|
||||
const done = engine.runOrchestrator(
|
||||
async (ctx) => {
|
||||
try {
|
||||
await ctx.wait((m) => m.type === 'tap');
|
||||
log.push('resolved');
|
||||
} catch {
|
||||
log.push('aborted');
|
||||
}
|
||||
},
|
||||
controller.signal,
|
||||
);
|
||||
|
||||
controller.abort();
|
||||
await done;
|
||||
expect(log).toEqual(['aborted']);
|
||||
});
|
||||
});
|
||||
@@ -1,53 +1,104 @@
|
||||
/**
|
||||
* The engine: the message bus that ties the queue, triggers, orchestrators,
|
||||
* and the command host together.
|
||||
* The engine: the message bus that ties the queue, triggers, and handlers
|
||||
* together.
|
||||
*
|
||||
* The engine is pure — no r3f, no React, no store. It defines the contract;
|
||||
* `@tts/tabletop` implements the `CommandHost` with the built-in commands that
|
||||
* mutate the tabletop store and drive the render layer. The engine never
|
||||
* imports tabletop.
|
||||
* `@tts/tabletop` registers the built-in commands (`move`, `focus`, `caption`,
|
||||
* ...) that mutate the tabletop store and drive the render layer. The engine
|
||||
* never imports tabletop.
|
||||
*/
|
||||
import { MessageQueue, type Message } from './message.js';
|
||||
import { MessageQueue, type CommandResult, type Message, type MessageHandler } from './message.js';
|
||||
import { TriggerRegistry, type Trigger } from './trigger.js';
|
||||
import { runOrchestrator, type Orchestrator } from './orchestrator.js';
|
||||
import { runOrchestrator } from './orchestrator.js';
|
||||
import { type Command, type Orchestrator, type RunContext } from './run.js';
|
||||
|
||||
/**
|
||||
* The command host: how a command handler registers with the bus. `@tts/tabletop`
|
||||
* implements this with the built-in commands (`move`, `focus`, `caption`, ...).
|
||||
* The engine dispatches each message to the host's registered handler for its
|
||||
* type; the host runs the command (the single mutation path) and emits the
|
||||
* `type:done` result message.
|
||||
*/
|
||||
export interface CommandHost {
|
||||
/** Register a handler for a message type. Returns an unsubscribe. */
|
||||
on(type: string, handler: (msg: Message) => void): () => void;
|
||||
/** Dispatch a message to the registered handler for its type, if any. */
|
||||
dispatch(msg: Message): void;
|
||||
}
|
||||
|
||||
/**
|
||||
* The engine. `enqueue` adds a message; `tick` drains the queue and processes
|
||||
* it through the command host and triggers. Orchestrators run against the same
|
||||
* queue and suspend on `wait` until a matching message is processed.
|
||||
* The engine. `enqueue` adds a message; `tick` drains the queue and dispatches
|
||||
* each message to the handlers registered for its `type`, then to matching
|
||||
* triggers. Orchestrators run against the same queue and suspend on `wait`
|
||||
* until a matching message is processed.
|
||||
*/
|
||||
export class Engine {
|
||||
private queue = new MessageQueue();
|
||||
private triggers = new TriggerRegistry();
|
||||
private host: CommandHost;
|
||||
private handlers = new Map<string, MessageHandler[]>();
|
||||
|
||||
constructor(host: CommandHost) {
|
||||
this.host = host;
|
||||
// Every message goes to the command host first (it runs the command), then
|
||||
// to triggers (they react to the message, including the `type:done` the
|
||||
// host emits). Emissions from either land on the next tick.
|
||||
constructor() {
|
||||
// Every message goes to the handlers for its type (they run commands), then
|
||||
// to triggers (they react to the message, including the `:done` a command
|
||||
// emits). Emissions from either land on the next tick.
|
||||
this.queue.on((msg) => {
|
||||
this.host.dispatch(msg);
|
||||
for (const handler of this.handlers.get(msg.type) ?? []) handler(msg);
|
||||
for (const t of this.triggers.match(msg)) {
|
||||
for (const emit of t.emit) this.queue.enqueue(emit);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** Register a handler for a message type. Returns an unsubscribe. */
|
||||
on(type: string, handler: MessageHandler): () => void {
|
||||
const list = this.handlers.get(type) ?? [];
|
||||
list.push(handler);
|
||||
this.handlers.set(type, list);
|
||||
return () => {
|
||||
const cur = this.handlers.get(type);
|
||||
if (!cur) return;
|
||||
const next = cur.filter((h) => h !== handler);
|
||||
if (next.length) this.handlers.set(type, next);
|
||||
else this.handlers.delete(type);
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a command. The engine builds a `RunContext` from the message,
|
||||
* runs the command, and emits its result — `:done` on resolve, `:cancel` on
|
||||
* abort, `:error` on throw.
|
||||
*/
|
||||
runCommand<Name extends string, Args, Result>(
|
||||
type: Name,
|
||||
command: Command<Args, Result>,
|
||||
): () => void {
|
||||
return this.on(type, (msg) => {
|
||||
const controller = new AbortController();
|
||||
const ctx: RunContext = {
|
||||
signal: controller.signal,
|
||||
emit: (m) => this.queue.enqueue(m),
|
||||
wait: (pred) =>
|
||||
new Promise<Message>((resolve, reject) => {
|
||||
if (controller.signal.aborted) {
|
||||
reject(new Error('aborted'));
|
||||
return;
|
||||
}
|
||||
const off = this.on('*', (m) => {
|
||||
if (pred(m)) {
|
||||
off();
|
||||
resolve(m);
|
||||
}
|
||||
});
|
||||
controller.signal.addEventListener('abort', () => {
|
||||
off();
|
||||
reject(new Error('aborted'));
|
||||
}, { once: true });
|
||||
}),
|
||||
enableTrigger: (type, id) => this.triggers.enable(type, id),
|
||||
disableTrigger: (type, id) => this.triggers.disable(type, id),
|
||||
};
|
||||
const args = msg.data as Args;
|
||||
command({ ...ctx, args })
|
||||
.then((result) => this.queue.enqueue({ type: `${type}:done`, data: result } as CommandResult<Name, Result>))
|
||||
.catch((err: unknown) => {
|
||||
if (controller.signal.aborted) {
|
||||
this.queue.enqueue({ type: `${type}:cancel` } as CommandResult<Name, Result>);
|
||||
} else {
|
||||
this.queue.enqueue({
|
||||
type: `${type}:error`,
|
||||
error: err instanceof Error ? err : new Error(String(err)),
|
||||
} as CommandResult<Name, Result>);
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
enqueue(msg: Message): void {
|
||||
this.queue.enqueue(msg);
|
||||
}
|
||||
@@ -74,19 +125,20 @@ export class Engine {
|
||||
}
|
||||
|
||||
/** Run an orchestrator against this engine's queue. */
|
||||
runOrchestrator(o: Orchestrator): Promise<void> {
|
||||
runOrchestrator(o: Orchestrator, signal?: AbortSignal): Promise<void> {
|
||||
return runOrchestrator(
|
||||
o,
|
||||
(msg) => this.queue.enqueue(msg),
|
||||
(handler) => this.queue.on(handler),
|
||||
(type, id) => this.triggers.enable(type, id),
|
||||
(type, id) => this.triggers.disable(type, id),
|
||||
signal,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
export type { Message, MessageHandler } from './message.js';
|
||||
export type { Message, MessageHandler, CommandMessage, CommandResult } from './message.js';
|
||||
export type { Trigger } from './trigger.js';
|
||||
export { triggerMatches, TriggerRegistry } from './trigger.js';
|
||||
export type { Orchestrator, OrchestratorContext } from './orchestrator.js';
|
||||
export type { Orchestrator, RunContext, Command } from './run.js';
|
||||
export { runOrchestrator } from './orchestrator.js';
|
||||
@@ -1,4 +1,11 @@
|
||||
export { MessageQueue, type Message, type MessageHandler } from './message.js';
|
||||
export {
|
||||
MessageQueue,
|
||||
type Message,
|
||||
type MessageHandler,
|
||||
type CommandMessage,
|
||||
type CommandResult,
|
||||
} from './message.js';
|
||||
export { TriggerRegistry, triggerMatches, type Trigger } from './trigger.js';
|
||||
export { runOrchestrator, type Orchestrator, type OrchestratorContext } from './orchestrator.js';
|
||||
export { Engine, type CommandHost } from './engine.js';
|
||||
export { runOrchestrator } from './orchestrator.js';
|
||||
export type { Orchestrator, RunContext, Command } from './run.js';
|
||||
export { Engine } from './engine.js';
|
||||
@@ -2,10 +2,13 @@
|
||||
* The message model and the queue that serializes it.
|
||||
*
|
||||
* A message is both an event (something happened) and an intent (something
|
||||
* should happen). It is identified by `type` (and optionally `id`), matching
|
||||
* the format's `type#id` convention. A `move` message both runs the move
|
||||
* command and is observable as an event; the command's completion is itself a
|
||||
* message (`move:done`), which is what triggers match and orchestrators await.
|
||||
* should happen). It is identified by `type`, matching the format's `type#id`
|
||||
* convention. A `move` message both runs the move command and is observable as
|
||||
* an event; the command's completion is itself a message (`move:done`), which
|
||||
* is what triggers match and orchestrators await.
|
||||
*
|
||||
* Messages are a discriminated union on `type`. The engine defines the generic
|
||||
* shapes; the host's concrete union extends them with its own command types.
|
||||
*
|
||||
* Messages are not processed inline. They are enqueued and handled on the next
|
||||
* `tick()`. This kills reentrancy (a handler cannot cause unbounded
|
||||
@@ -13,12 +16,34 @@
|
||||
* deterministic frame. The engine is pure — it has no render loop — so the
|
||||
* host calls `tick()` (a `useFrame` in `@tts/tabletop`, manually in tests).
|
||||
*/
|
||||
|
||||
/** A command message names a command to run. Its handler is the command. */
|
||||
export interface CommandMessage<Name extends string, Args> {
|
||||
type: Name;
|
||||
data: Args;
|
||||
}
|
||||
|
||||
/**
|
||||
* A command's result, discriminated on the type suffix. `:done` on resolve,
|
||||
* `:cancel` on abort (superseded, skipped, surface disabled), `:error` on
|
||||
* throw. A trigger matching `move:done` does not fire on a cancel.
|
||||
*/
|
||||
export type CommandResult<Name extends string, R = void> =
|
||||
| { type: `${Name}:done`; data: R }
|
||||
| { type: `${Name}:cancel` }
|
||||
| { type: `${Name}:error`; error: Error };
|
||||
|
||||
/**
|
||||
* The base message type. The engine is host-agnostic, so this is a permissive
|
||||
* structural type; the host defines a concrete discriminated union on `type`
|
||||
* that extends it with its own command and interaction messages.
|
||||
*/
|
||||
export interface Message {
|
||||
type: string;
|
||||
/** Command-specific payload. */
|
||||
data?: unknown;
|
||||
/** Optional identity, for matching and dedup. */
|
||||
id?: string;
|
||||
/** Present on `:error` result messages. */
|
||||
error?: Error;
|
||||
}
|
||||
|
||||
/** A handler consumes a message and may emit new ones. */
|
||||
|
||||
@@ -3,7 +3,8 @@
|
||||
*
|
||||
* An orchestrator is the code counterpart to a trigger: an async function that
|
||||
* emits messages and awaits matching ones. It is a proper TS module, declared
|
||||
* per folder as `main.ts`, unique per folder like `package.yaml`.
|
||||
* per folder as `main.ts`, unique per folder like `package.yaml`, exported as a
|
||||
* default async function.
|
||||
*
|
||||
* An orchestrator is a long-running command: it awaits events instead of
|
||||
* resolving immediately, so it inherits the run-context machinery (supersede
|
||||
@@ -11,25 +12,12 @@
|
||||
* — declaration is data, activation is code.
|
||||
*/
|
||||
import type { Message } from './message.js';
|
||||
|
||||
/** The context an orchestrator runs against. */
|
||||
export interface OrchestratorContext {
|
||||
/** Emit a message onto the queue. */
|
||||
emit(msg: Message): void;
|
||||
/** Await the next message matching `pred`. */
|
||||
wait(pred: (msg: Message) => boolean): Promise<Message>;
|
||||
/** Enable/disable a trigger by `type#id`. */
|
||||
enableTrigger(type: string, id?: string): void;
|
||||
disableTrigger(type: string, id?: string): void;
|
||||
}
|
||||
|
||||
/** An orchestrator: an async function that emits and awaits messages. */
|
||||
export type Orchestrator = (ctx: OrchestratorContext) => Promise<void>;
|
||||
import type { Orchestrator, RunContext } from './run.js';
|
||||
|
||||
/**
|
||||
* Run an orchestrator against a queue. `emit` enqueues; `wait` suspends until
|
||||
* a matching message is processed during a `tick()`. Returns a promise that
|
||||
* resolves when the orchestrator completes.
|
||||
* a matching message is processed during a `tick()`, rejecting on abort.
|
||||
* Returns a promise that resolves when the orchestrator completes.
|
||||
*/
|
||||
export function runOrchestrator(
|
||||
orchestrator: Orchestrator,
|
||||
@@ -37,8 +25,18 @@ export function runOrchestrator(
|
||||
on: (handler: (msg: Message) => void) => () => void,
|
||||
enableTrigger: (type: string, id?: string) => void,
|
||||
disableTrigger: (type: string, id?: string) => void,
|
||||
signal?: AbortSignal,
|
||||
): Promise<void> {
|
||||
const pending: Array<{ pred: (msg: Message) => boolean; resolve: (m: Message) => void }> = [];
|
||||
const controller = new AbortController();
|
||||
const pending: Array<{ pred: (msg: Message) => boolean; resolve: (m: Message) => void; reject: (e: Error) => void }> = [];
|
||||
const abort = () => {
|
||||
controller.abort();
|
||||
// Reject every pending wait so the orchestrator unwinds on cancel.
|
||||
for (const p of pending.splice(0)) p.reject(new Error('aborted'));
|
||||
};
|
||||
if (signal?.aborted) abort();
|
||||
else signal?.addEventListener('abort', abort, { once: true });
|
||||
|
||||
const unsubscribe = on((msg) => {
|
||||
for (let i = 0; i < pending.length; i++) {
|
||||
const p = pending[i]!;
|
||||
@@ -50,15 +48,23 @@ export function runOrchestrator(
|
||||
}
|
||||
});
|
||||
|
||||
const ctx: OrchestratorContext = {
|
||||
const ctx: RunContext = {
|
||||
signal: controller.signal,
|
||||
emit,
|
||||
wait: (pred) =>
|
||||
new Promise<Message>((resolve) => {
|
||||
pending.push({ pred, resolve });
|
||||
new Promise<Message>((resolve, reject) => {
|
||||
if (controller.signal.aborted) {
|
||||
reject(new Error('aborted'));
|
||||
return;
|
||||
}
|
||||
pending.push({ pred, resolve, reject });
|
||||
}),
|
||||
enableTrigger,
|
||||
disableTrigger,
|
||||
};
|
||||
|
||||
return orchestrator(ctx).finally(unsubscribe);
|
||||
return orchestrator(ctx).finally(() => {
|
||||
unsubscribe();
|
||||
controller.abort();
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
/**
|
||||
* The run context and the unified command/orchestrator shape.
|
||||
*
|
||||
* A command and an orchestrator are the same thing: an async function taking a
|
||||
* `RunContext`. A command returns a result and is awaited by the engine, which
|
||||
* emits `:done`/`:cancel`/`:error`; an orchestrator returns `void` and is never
|
||||
* awaited by a parent. Cancellation is an `AbortSignal` — a superseded command
|
||||
* or disabled surface aborts it, `wait` rejects on abort, and the `:cancel`
|
||||
* result is emitted. Errors are thrown: a command that throws emits `:error`.
|
||||
*/
|
||||
import type { Message } from './message.js';
|
||||
|
||||
/** The handle every handler runs against. */
|
||||
export interface RunContext {
|
||||
/** Cancellation: superseded, skipped, surface disabled. */
|
||||
signal: AbortSignal;
|
||||
/** Emit a message onto the queue. */
|
||||
emit(msg: Message): void;
|
||||
/** Await the next message matching `pred`. Rejects on abort. */
|
||||
wait(pred: (m: Message) => boolean): Promise<Message>;
|
||||
/** Enable/disable a trigger by `type#id`. */
|
||||
enableTrigger(type: string, id?: string): void;
|
||||
disableTrigger(type: string, id?: string): void;
|
||||
}
|
||||
|
||||
/** A command: an async function taking the context plus its args. */
|
||||
export type Command<Args, Result> = (ctx: RunContext & { args: Args }) => Promise<Result>;
|
||||
|
||||
/** An orchestrator: an async function that emits and awaits messages. */
|
||||
export type Orchestrator = (ctx: RunContext) => Promise<void>;
|
||||
Reference in New Issue
Block a user