novy-syndicate/actor.ts

import { IdentitySet, Value } from 'preserves';

export type Assertion = Value<Ref<Entity>>;

export type Handle = any;

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 const sync = Symbol('sync');
export const synced = Symbol('synced');

export interface Entity {
    [assert](turn: Turn, assertion: Assertion, handle: Handle): void;
    [retract](turn: Turn, handle: Handle): void;
    [message](turn: Turn, message: Assertion): void;
}

export interface SyncTarget {
    [sync](turn: Turn, source: Ref<SyncSource>): void;
}

export interface SyncSource {
    [synced](turn: Turn): void;
}

export class Ref<T = Entity> {
    readonly actor: Actor;
    readonly target: T;
    constructor(actor: Actor, target: T) {
        this.actor = actor;
        this.target = target;
    }
}

export type OutboundMap = Map<Handle, [Ref, Assertion]>;

export class Actor implements SyncTarget {
    readonly outbound: OutboundMap;
    exitReason: ExitReason = null;

    constructor(initialAssertions: OutboundMap = new Map()) {
        this.outbound = initialAssertions;
    }

    get alive(): boolean {
        return this.exitReason === null;
    }

    stop() {
        this.terminateWith({ ok: true });
    }

    terminateWith(reason: Exclude<ExitReason, null>) {
        if (this.alive) {
            this.exitReason = reason;
            Turn.for(this, t => this.outbound.forEach(([peer, _a], h) => t._retract(peer, h)));
        }
    }

    ref<T>(t: T): Ref<T> {
        return new Ref(this, t);
    }

    execute(proc: () => void): void {
        queueMicrotask(() => {
            if (this.alive) {
                try {
                    proc();
                } catch (err) {
                    this.terminateWith({ ok: false, err });
                }
            }
        });
    }

    [sync](t: Turn, source: Ref<SyncSource>): void {
        t.synced(source);
    }
}

let nextHandle = 0;

type LocalAction = (t: Turn) => void;

export class Turn {
    readonly actor: Actor | null; // whose turn it is to act during this Turn
    readonly queues: Map<Actor, LocalAction[]> = new Map();
    readonly localActions: Array<LocalAction> = [];
    completed = false;

    static for(actor: Actor | null, f: (t: Turn) => void): void {
        const t = new Turn(actor);
        f(t);
        t.complete();
    }

    private constructor(actor: Actor | null) {
        this.actor = actor;
    }

    _ensureActor(what: string): Actor {
        if (this.actor === null) {
            throw new Error(`Cannot ${what} from non-Actor context`);
        }
        return this.actor;
    }

    spawn(bootProc: (t: Turn) => void, initialAssertions?: IdentitySet<Handle>): void {
        if ((initialAssertions !== void 0) && (initialAssertions.size > 0)) {
            this._ensureActor("spawn with initialAssertions");
        }
        this.localActions.push(() => {
            const transferred = this.actor === null
                ? void 0
                : extractFromMap(this.actor.outbound, initialAssertions);
            const child = new Actor(transferred);
            child.execute(() => Turn.for(child, bootProc));
        });
    }

    assert(location: Ref, assertion: Assertion): Handle {
        this._ensureActor("assert");
        const h = nextHandle++;
        this.enqueue(location.actor, t => {
            this.actor!.outbound.set(h, [location, assertion]);
            location.target[assert](t, assertion, h);
        });
        return h;
    }

    retract(h: Handle): void {
        this._retract(this._ensureActor("retract").outbound.get(h)![0], h);
    }

    _retract(location: Ref, handle: Handle): void {
        this.enqueue(location.actor, t => {
            this.actor!.outbound.delete(handle);
            location.target[retract](t, handle);
        });
    }

    sync(location: Ref<SyncTarget>): Promise<Turn> {
        return new Promise(resolve => {
            const k = this._ensureActor("sync").ref({ [synced]: resolve });
            this.enqueue(location.actor, t => location.target[sync](t, k));
        });
    }

    synced(syncable: Ref<SyncSource>): void {
        this.enqueue(syncable.actor, t => syncable.target[synced](t));
    }

    message(location: Ref, assertion: Assertion): void {
        this.enqueue(location.actor, t => location.target[message](t, assertion));
    }

    enqueue(actor: Actor, a: LocalAction): void {
        let queue = this.queues.get(actor);
        if (queue === void 0) {
            queue = [];
            this.queues.set(actor, queue);
        }
        queue.push(a);
    }

    complete(): void {
        if (this.completed) {
            throw new Error("Reuse of completed Turn!");
        }
        this.completed = true;
        this.queues.forEach((queue, actor) =>
            actor.execute(() => queue.forEach(f => Turn.for(actor, f))));
        this.localActions.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();
    if (keys !== void 0) {
        keys.forEach(key => {
            const value = map.get(key);
            if (value !== void 0) {
                map.delete(key);
                result.set(key, value);
            }
        });
    }
    return result;
}