novy-syndicate/main.ts

import { Actor, Assertion, Entity, Handle, Ref, Turn, assert, message, retract } from './actor.js';
import { Dictionary, IdentityMap, is, Record } from 'preserves';
import { Bag, ChangeDescription } from './bag';

const Observe = Record.makeConstructor<Ref<Entity>>('Observe', ['label', 'observer']);

function makeDataspace(): Entity {
    const handleMap: IdentityMap<Handle, Record<Ref<Entity>>> = new IdentityMap();
    const assertions = new Bag<Ref<Entity>>();
    const subscriptions: Dictionary<Map<Ref<Entity>, Dictionary<Handle>>> = new Dictionary();

    return {
        [assert](turn: Turn, rec: Assertion, handle: Handle): void {
            if (!Record.isRecord<Ref<Entity>>(rec)) return;
            handleMap.set(handle, rec);
            if (assertions.change(rec, +1) !== ChangeDescription.ABSENT_TO_PRESENT) return;
            if (Observe.isClassOf(rec)) {
                const label = Observe._.label(rec)!;
                const observer = Observe._.observer(rec) as Ref<Entity>;
                const seen = new Dictionary<Handle>();
                if (!subscriptions.has(label)) subscriptions.set(label, new Map());
                subscriptions.get(label)!.set(observer, seen);
                assertions.forEach((_count, prev) =>
                    is((prev as Record<Ref<Entity>>).label, label)
                    && seen.set(prev, turn.assert(observer, prev)));
            }
            subscriptions.get(rec.label)?.forEach((seen, peer) =>
                seen.has(rec) || seen.set(rec, turn.assert(peer, rec)));
        },

        [retract](turn: Turn, upstreamHandle: Handle): void {
            const rec = handleMap.get(upstreamHandle);
            if (rec === void 0) return;
            handleMap.delete(upstreamHandle);
            if (assertions.change(rec, -1) !== ChangeDescription.PRESENT_TO_ABSENT) return;
            subscriptions.get(rec.label)?.forEach((seen, _peer) => {
                const downstreamHandle = seen.get(rec);
                if (downstreamHandle !== void 0) {
                    turn.retract(downstreamHandle);
                    seen.delete(rec);
                }
            });
            if (Observe.isClassOf(rec)) {
                let peerMap = subscriptions.get(Observe._.label(rec)!)!;
                peerMap.delete(Observe._.observer(rec) as Ref<Entity>);
                if (peerMap.size === 0) subscriptions.delete(Observe._.label(rec)!);
            }
        },

        [message](turn: Turn, rec: Assertion): void {
            if (!Record.isRecord<Ref<Entity>>(rec)) return;
            subscriptions.get(rec.label)?.forEach((_seen, peer) => turn.message(peer, rec));
        }
    };
}

const BoxState = Record.makeConstructor<Ref<Entity>>('BoxState', ['value']);
const SetBox = Record.makeConstructor<Ref<Entity>>('SetBox', ['newValue']);

let startTime = Date.now();
let prevValue = 0;
Turn.for(null, async (t: Turn) => {
    const ds = new Ref(new Actor(), makeDataspace());

    // Box
    t.spawn(t => {
        console.log('Spawning Box');
        let valueHandle: Handle | undefined;
        function setValue(t: Turn, value: number) {
            valueHandle = t.replace(ds, valueHandle, BoxState(value));
        }
        setValue(t, 0);
        t.assert(ds, Observe(SetBox.constructorInfo.label, t.ref({
            [message](t: Turn, [newValue]: [number]): void {
                if (newValue % 25000 === 0) {
                    const endTime = Date.now();
                    const delta = (endTime - startTime) / 1000.0;
                    const count = newValue - prevValue;
                    prevValue = newValue;
                    startTime = endTime;
                    console.log(`Box: got ${newValue} (${count / delta} Hz)`);
                }
                if (newValue === 280000) t.quit();
                setValue(t, newValue);
            }
        })));
    });

    // Client
    t.spawn(t => {
        console.log('Spawning Client');
        let count = 0;
        t.assert(ds, Observe(BoxState.constructorInfo.label, t.ref({
            [assert](t: Turn, [currentValue]: [number]): void {
                // console.log(`Client: got ${currentValue}`);
                if (currentValue === 300000) {
                    console.log(`Client: quitting at limit`);
                    t.quit();
                } else {
                    t.message(ds, SetBox(currentValue + 1));
                }
            }
        })));
        t.assert(ds, Observe(BoxState.constructorInfo.label, t.ref({
            [assert](_t: Turn, _assertion: Assertion): void { count++; },
            [retract](t: Turn, _handle: Handle) {
                if (--count === 0) {
                    console.log('Client: detected box termination');
                    t.quit();
                }
            },
        })));
    });
});