A module's state through the runtime, in TypeScript and Go (novox/hq ADR 0201)
state(name) / stdio.State(name): get, put, delete, keys and watch over the runtime's mesh/state verbs. A watch's handler is registered under the id it is asked with, so its current values — which arrive before the watch is answered — and its changes reach it and no other watch.
This commit is contained in:
+38
-2
@@ -15,6 +15,7 @@
|
||||
import type { ToolDefinition } from "../contracts/index.js";
|
||||
import { collectTools } from "../tools/index.js";
|
||||
import { useBroker, type Broker, type Envelope } from "../messaging/index.js";
|
||||
import { useStateChannel, type StateChange } from "../state/index.js";
|
||||
|
||||
/** The protocol version this speaks; the runtime says the same. */
|
||||
export const PROTOCOL = "2025-03-26";
|
||||
@@ -63,12 +64,30 @@ function startSession(): Session {
|
||||
};
|
||||
let nextOut = 1;
|
||||
const awaiting = new Map<string, { resolve: (v: unknown) => void; reject: (e: Error) => void }>();
|
||||
const ask = (method: string, params: unknown): Promise<unknown> =>
|
||||
const askAs = (id: string, method: string, params: unknown): Promise<unknown> =>
|
||||
new Promise((resolve, reject) => {
|
||||
const id = `bundle-${nextOut++}`;
|
||||
awaiting.set(id, { resolve, reject });
|
||||
say({ jsonrpc: "2.0", id, method, params });
|
||||
});
|
||||
const ask = (method: string, params: unknown): Promise<unknown> => askAs(`bundle-${nextOut++}`, method, params);
|
||||
|
||||
// The module's state (ADR 0201): each watch's changes arrive as `mesh/state` requests naming the
|
||||
// watch by the id it was asked with, so its handler is registered under that id before it is asked
|
||||
// — the current values arrive before the watch itself is answered.
|
||||
const watches = new Map<string, (change: StateChange<unknown>) => Promise<void>>();
|
||||
useStateChannel({
|
||||
ask,
|
||||
watch: async (params, deliver) => {
|
||||
const id = `bundle-${nextOut++}`;
|
||||
watches.set(id, deliver);
|
||||
try {
|
||||
await askAs(id, "mesh/state.watch", params);
|
||||
} catch (err) {
|
||||
watches.delete(id);
|
||||
throw err;
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
// What the module's code reacts to: every handler, matched here against each event the runtime
|
||||
// delivers. The runtime binds the module's one durable consumer and hands every message it reads to
|
||||
@@ -156,6 +175,23 @@ function startSession(): Session {
|
||||
}
|
||||
return;
|
||||
}
|
||||
case "mesh/state": {
|
||||
// One change to a state this bundle watches (ADR 0201), answered once its handler has taken
|
||||
// it. The runtime delivers a watch's changes one at a time, in order.
|
||||
const p = request.params as { watch?: string; change?: StateChange<unknown> } | undefined;
|
||||
const deliver = p?.watch !== undefined ? watches.get(String(p.watch)) : undefined;
|
||||
if (!deliver || !p?.change) {
|
||||
say({ jsonrpc: "2.0", id: request.id, error: { code: -32602, message: "mesh/state names no watch of this bundle" } });
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await deliver(p.change);
|
||||
say({ jsonrpc: "2.0", id: request.id, result: {} });
|
||||
} catch (err) {
|
||||
say({ jsonrpc: "2.0", id: request.id, error: { code: -32000, message: err instanceof Error ? err.message : String(err) } });
|
||||
}
|
||||
return;
|
||||
}
|
||||
default:
|
||||
if (!notification) say({ jsonrpc: "2.0", id: request.id, error: { code: -32601, message: `no ${request.method}` } });
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user