novy-syndicate/actor.ts

149 lines
4.2 KiB
TypeScript

import { IdentitySet, Value } from 'preserves';
export type Assertion = Value<Ref<Entity>>;
export type Handle = number;
export type ExitReason = null | { ok: true } | { ok: false, err: Error };
export const assert = Symbol('assert');
export const retract = Symbol('retract');
export const message = Symbol('message');
export interface Entity {
[assert]?(turn: Turn, assertion: Assertion, handle: Handle): void;
[retract]?(turn: Turn, handle: Handle): void;
[message]?(turn: Turn, message: Assertion): void;
}
export class Ref<T> {
readonly actor: Actor;
readonly target: T;
constructor(actor: Actor, target: T) {
this.actor = actor;
this.target = target;
}
sync(turn: Turn, syncable: Ref<LocalAction>) {
turn.enqueue(syncable.actor, t => syncable.target(t));
}
}
export type OutboundMap = Map<Handle, Ref<Entity>>;
export class Actor {
readonly outbound: OutboundMap;
exitReason: ExitReason = null;
constructor(initialAssertions: OutboundMap = new Map()) {
this.outbound = initialAssertions;
}
terminateWith(t: Turn, reason: Exclude<ExitReason, null>) {
if (this.exitReason !== null) return;
this.exitReason = reason;
this.outbound.forEach((peer, h) => t._retract(peer, h));
}
execute(proc: () => void): void {
queueMicrotask(() => {
if (this.exitReason !== null) return;
try {
proc();
} catch (err) {
console.error(Actor, err);
Turn.for(this, t => this.terminateWith(t, { ok: false, err }));
}
});
}
}
let nextHandle = 0;
type LocalAction = (t: Turn) => void;
export class Turn {
readonly actor: Actor;
readonly queues: Map<Actor, LocalAction[]> = new Map();
readonly tasks: Array<LocalAction> = [];
static for(actor: Actor, f: (t: Turn) => void): void {
const t = new Turn(actor);
f(t);
t.complete();
}
private constructor(actor: Actor) {
this.actor = actor;
}
ref<T>(t: T): Ref<T> {
return new Ref(this.actor, t);
}
spawn(bootProc: (t: Turn) => void, initialAssertions = new IdentitySet<Handle>()): void {
this.tasks.push(() => {
const child = new Actor(extractFromMap(this.actor.outbound, initialAssertions));
child.execute(() => Turn.for(child, bootProc));
});
}
quit(): void {
this.tasks.push(t => this.actor.terminateWith(t, { ok: true }));
}
assert(location: Ref<Entity>, assertion: Assertion): Handle {
const h = nextHandle++;
this.enqueue(location.actor, t => {
this.actor.outbound.set(h, location);
location.target[assert]?.(t, assertion, h);
});
return h;
}
retract(h: Handle): void {
this._retract(this.actor.outbound.get(h)!, h);
}
replace(location: Ref<Entity>, h: Handle | undefined, assertion: Assertion): Handle {
const newHandle = this.assert(location, assertion);
if (h !== void 0) this.retract(h);
return newHandle;
}
_retract(location: Ref<Entity>, handle: Handle): void {
this.enqueue(location.actor, t => {
this.actor.outbound.delete(handle);
location.target[retract]?.(t, handle);
});
}
sync(location: Ref<any>): Promise<Turn> {
return new Promise(resolve =>
this.enqueue(location.actor, t => location.sync(t, this.ref(resolve))));
}
message(location: Ref<Entity>, assertion: Assertion): void {
this.enqueue(location.actor, t => location.target[message]?.(t, assertion));
}
enqueue(actor: Actor, a: LocalAction): void {
this.queues.get(actor)?.push(a) ?? this.queues.set(actor, [a]);
}
private complete(): void {
this.queues.forEach((q, a) => a.execute(() => q.forEach(f => Turn.for(a, f))));
this.tasks.length && queueMicrotask(() => this.tasks.forEach(f => Turn.for(this.actor, f)));
}
}
function extractFromMap<K, V>(map: Map<K, V>, keys: IdentitySet<K>): Map<K, V> {
const result: Map<K, V> = new Map();
keys.forEach(key => {
if (map.has(key)) result.set(key, map.get(key)!);
map.delete(key);
});
return result;
}