index.ts raw
1 // anneal - CSP/Actor building blocks for TypeScript
2 //
3 // Five channel types covering every actor communication pattern:
4 // Func<Req, Resp> request -> response (function call)
5 // Query<Resp> void -> response (getter)
6 // Proc<Req> request -> done (setter/command)
7 // Signal void -> done (trigger/reset)
8 // Inbox<T> fire-and-forget (logging/notifications)
9 //
10 // Lifecycle manages the stop/done shutdown protocol.
11 // select() and loop() provide the actor's message dispatch.
12
13 // ---------------------------------------------------------------------------
14 // Sentinel
15 // ---------------------------------------------------------------------------
16
17 /** Return from a handler in select/loop to terminate the loop. */
18 export const STOP: unique symbol = Symbol('anneal.stop');
19 export type Stop = typeof STOP;
20
21 // ---------------------------------------------------------------------------
22 // Mailbox (internal)
23 // ---------------------------------------------------------------------------
24
25 class Mailbox<T> {
26 private q: T[] = [];
27 private waker: (() => void) | null = null;
28
29 post(v: T): void {
30 this.q.push(v);
31 if (this.waker) {
32 const fn = this.waker;
33 this.waker = null;
34 fn();
35 }
36 }
37
38 get pending(): boolean { return this.q.length > 0; }
39 get length(): number { return this.q.length; }
40
41 take(): T { return this.q.shift()!; }
42
43 wake(fn: () => void): void { this.waker = fn; }
44 unwake(): void { this.waker = null; }
45 }
46
47 // ---------------------------------------------------------------------------
48 // Case
49 // ---------------------------------------------------------------------------
50
51 /** A select case produced by a channel's .on() method. */
52 export interface Case {
53 /** @internal */ readonly _mb: unknown;
54 /** @internal */ readonly _fn: (msg: any) => unknown;
55 }
56
57 function makeCase<T>(mb: Mailbox<T>, fn: (msg: T) => void | Stop | Promise<void | Stop>): Case {
58 return { _mb: mb, _fn: fn };
59 }
60
61 function caseMb(c: Case): Mailbox<any> {
62 return c._mb as Mailbox<any>;
63 }
64
65 // ---------------------------------------------------------------------------
66 // select / loop
67 // ---------------------------------------------------------------------------
68
69 /**
70 * Wait for one message from any of the given cases and run its
71 * handler. Cases are checked in order - first pending wins.
72 */
73 export async function select(...cases: Case[]): Promise<unknown> {
74 for (const c of cases) {
75 const mb = caseMb(c);
76 if (mb.pending) return c._fn(mb.take());
77 }
78 return new Promise<unknown>(resolve => {
79 let fired = false;
80 const cleanup = () => { for (const c of cases) caseMb(c).unwake(); };
81 for (const c of cases) {
82 const mb = caseMb(c);
83 mb.wake(() => {
84 if (fired) return;
85 fired = true;
86 cleanup();
87 resolve(c._fn(mb.take()));
88 });
89 }
90 });
91 }
92
93 /** Run select in a loop until a handler returns STOP. */
94 export async function loop(...cases: Case[]): Promise<void> {
95 for (;;) {
96 const r = await select(...cases);
97 if (r === STOP) return;
98 }
99 }
100
101 // ---------------------------------------------------------------------------
102 // Func<Req, Resp>
103 // ---------------------------------------------------------------------------
104
105 /** Message received by the actor from a Func channel. */
106 export interface FuncMsg<Req, Resp> {
107 readonly req: Req;
108 /** Send the response back to the caller. Call exactly once. */
109 reply(resp: Resp): void;
110 }
111
112 /**
113 * Synchronous request-response channel. The caller sends a request
114 * and awaits the actor's reply.
115 */
116 export class Func<Req, Resp> {
117 private mb = new Mailbox<FuncMsg<Req, Resp>>();
118
119 /** Send req to the actor and await the response. */
120 call(req: Req, signal?: AbortSignal): Promise<Resp> {
121 if (signal?.aborted) return Promise.reject(signal.reason);
122 return new Promise<Resp>((resolve, reject) => {
123 let done = false;
124 const onAbort = signal ? () => {
125 if (!done) { done = true; reject(signal.reason); }
126 } : undefined;
127 this.mb.post({
128 req,
129 reply: (resp: Resp) => {
130 if (!done) {
131 done = true;
132 if (onAbort) signal!.removeEventListener('abort', onAbort);
133 resolve(resp);
134 }
135 },
136 });
137 if (onAbort) signal!.addEventListener('abort', onAbort, { once: true });
138 });
139 }
140
141 /** Create a select case for this channel. */
142 on(handler: (msg: FuncMsg<Req, Resp>) => void | Stop | Promise<void | Stop>): Case {
143 return makeCase(this.mb, handler);
144 }
145 }
146
147 // ---------------------------------------------------------------------------
148 // Query<Resp>
149 // ---------------------------------------------------------------------------
150
151 /** Message received by the actor from a Query channel. */
152 export interface QueryMsg<Resp> {
153 /** Send the response back to the caller. Call exactly once. */
154 reply(resp: Resp): void;
155 }
156
157 /**
158 * Synchronous getter channel. The caller sends no data and awaits
159 * the actor's reply.
160 */
161 export class Query<Resp> {
162 private mb = new Mailbox<QueryMsg<Resp>>();
163
164 /** Query the actor and await the response. */
165 call(signal?: AbortSignal): Promise<Resp> {
166 if (signal?.aborted) return Promise.reject(signal.reason);
167 return new Promise<Resp>((resolve, reject) => {
168 let done = false;
169 const onAbort = signal ? () => {
170 if (!done) { done = true; reject(signal.reason); }
171 } : undefined;
172 this.mb.post({
173 reply: (resp: Resp) => {
174 if (!done) {
175 done = true;
176 if (onAbort) signal!.removeEventListener('abort', onAbort);
177 resolve(resp);
178 }
179 },
180 });
181 if (onAbort) signal!.addEventListener('abort', onAbort, { once: true });
182 });
183 }
184
185 /** Create a select case for this channel. */
186 on(handler: (msg: QueryMsg<Resp>) => void | Stop | Promise<void | Stop>): Case {
187 return makeCase(this.mb, handler);
188 }
189 }
190
191 // ---------------------------------------------------------------------------
192 // Proc<Req>
193 // ---------------------------------------------------------------------------
194
195 /** Message received by the actor from a Proc channel. */
196 export interface ProcMsg<Req> {
197 readonly req: Req;
198 /** Signal the caller that processing is complete. Call exactly once. */
199 done(): void;
200 }
201
202 /**
203 * Synchronous command channel. The caller sends a request and awaits
204 * confirmation that the actor has processed it.
205 */
206 export class Proc<Req> {
207 private mb = new Mailbox<ProcMsg<Req>>();
208
209 /** Send req to the actor and await completion. */
210 call(req: Req, signal?: AbortSignal): Promise<void> {
211 if (signal?.aborted) return Promise.reject(signal.reason);
212 return new Promise<void>((resolve, reject) => {
213 let settled = false;
214 const onAbort = signal ? () => {
215 if (!settled) { settled = true; reject(signal.reason); }
216 } : undefined;
217 this.mb.post({
218 req,
219 done: () => {
220 if (!settled) {
221 settled = true;
222 if (onAbort) signal!.removeEventListener('abort', onAbort);
223 resolve();
224 }
225 },
226 });
227 if (onAbort) signal!.addEventListener('abort', onAbort, { once: true });
228 });
229 }
230
231 /** Create a select case for this channel. */
232 on(handler: (msg: ProcMsg<Req>) => void | Stop | Promise<void | Stop>): Case {
233 return makeCase(this.mb, handler);
234 }
235 }
236
237 // ---------------------------------------------------------------------------
238 // Signal
239 // ---------------------------------------------------------------------------
240
241 /** Message received by the actor from a Signal channel. */
242 export interface SignalMsg {
243 /** Signal the caller that the trigger has been handled. Call exactly once. */
244 done(): void;
245 }
246
247 /**
248 * Synchronous trigger channel. The caller sends no data and awaits
249 * confirmation that the actor has handled the signal.
250 */
251 export class Signal {
252 private mb = new Mailbox<SignalMsg>();
253
254 /** Send the signal and await completion. */
255 call(signal?: AbortSignal): Promise<void> {
256 if (signal?.aborted) return Promise.reject(signal.reason);
257 return new Promise<void>((resolve, reject) => {
258 let settled = false;
259 const onAbort = signal ? () => {
260 if (!settled) { settled = true; reject(signal.reason); }
261 } : undefined;
262 this.mb.post({
263 done: () => {
264 if (!settled) {
265 settled = true;
266 if (onAbort) signal!.removeEventListener('abort', onAbort);
267 resolve();
268 }
269 },
270 });
271 if (onAbort) signal!.addEventListener('abort', onAbort, { once: true });
272 });
273 }
274
275 /** Create a select case for this channel. */
276 on(handler: (msg: SignalMsg) => void | Stop | Promise<void | Stop>): Case {
277 return makeCase(this.mb, handler);
278 }
279 }
280
281 // ---------------------------------------------------------------------------
282 // Inbox<T>
283 // ---------------------------------------------------------------------------
284
285 /**
286 * Buffered fire-and-forget channel. Senders do not wait for the
287 * actor to process the message.
288 */
289 export class Inbox<T> {
290 private mb = new Mailbox<T>();
291 private cap: number;
292
293 constructor(buffer: number) {
294 this.cap = Math.max(0, buffer);
295 }
296
297 /** Enqueue a message. Always succeeds. */
298 send(v: T): void {
299 this.mb.post(v);
300 }
301
302 /**
303 * Attempt to enqueue a message. Returns false if the buffer
304 * capacity has been reached (message dropped).
305 */
306 trySend(v: T): boolean {
307 if (this.cap > 0 && this.mb.length >= this.cap) return false;
308 this.mb.post(v);
309 return true;
310 }
311
312 /** Create a select case for this channel. */
313 on(handler: (msg: T) => void | Stop | Promise<void | Stop>): Case {
314 return makeCase(this.mb, handler);
315 }
316 }
317
318 // ---------------------------------------------------------------------------
319 // Lifecycle
320 // ---------------------------------------------------------------------------
321
322 /**
323 * Manages the shutdown handshake between an actor and its owner.
324 * The owner calls stop() to request shutdown and wait for completion.
325 * The actor uses .on() in its select loop to detect the stop signal.
326 */
327 export class Lifecycle {
328 private mb = new Mailbox<void>();
329 private resolvers: (() => void)[] = [];
330 private isDone = false;
331 private hasStopped = false;
332
333 /** Create a select case for the stop signal. */
334 on(handler: () => void | Stop | Promise<void | Stop>): Case {
335 return makeCase(this.mb, handler);
336 }
337
338 /** Promise that resolves when the actor has finished. */
339 get stopped(): Promise<void> {
340 if (this.isDone) return Promise.resolve();
341 return new Promise(r => this.resolvers.push(r));
342 }
343
344 /** Signal the actor to stop and wait for it to finish. */
345 stop(): Promise<void> {
346 if (!this.hasStopped) {
347 this.hasStopped = true;
348 this.mb.post(undefined);
349 }
350 return this.stopped;
351 }
352
353 /**
354 * Signal that the actor has finished. Called automatically by
355 * spawn - do not call directly.
356 */
357 markDone(): void {
358 if (!this.isDone) {
359 this.isDone = true;
360 for (const r of this.resolvers) r();
361 this.resolvers.length = 0;
362 }
363 }
364 }
365
366 // ---------------------------------------------------------------------------
367 // spawn
368 // ---------------------------------------------------------------------------
369
370 /**
371 * Start an actor function with lifecycle management. Calls
372 * lc.markDone() when the async function returns or throws.
373 */
374 export function spawn(lc: Lifecycle, fn: () => Promise<void>): void {
375 fn().finally(() => lc.markDone());
376 }
377