Factor out different kinds of IndexObserver

packages/core/src/runtime/dataspace.ts

@@ -1,9 +1,9 @@
 /// SPDX-License-Identifier: GPL-3.0-or-later
 /// SPDX-FileCopyrightText: Copyright © 2016-2023 Tony Garnock-Jones <tonyg@leastfixedpoint.com>
 
-import { IdentityMap } from '@preserves/core';
+import { IdentityMap, KeyedDictionary, stringify } from '@preserves/core';
-import { Index } from './skeleton.js';
+import { Index, IndexObserver } from './skeleton.js';
-import { Actor, Assertion, Entity, Facet, Handle, LocalAction, Ref, Turn } from './actor.js';
+import { Actor, AnyValue, Assertion, Entity, Facet, Handle, LocalAction, Ref, Turn } from './actor.js';
 import { Observe, toObserve } from '../gen/dataspace.js';
 import * as P from '../gen/dataspacePatterns.js';
 
@@ -19,6 +19,7 @@ export class Dataspace implements Partial<Entity> {
     readonly options: DataspaceOptions;
     readonly index = new Index();
     readonly handleMap = new IdentityMap<Handle, Assertion>();
+    readonly observerMap = new IdentityMap<Ref, IndexObserver>();
 
     constructor(options?: DataspaceOptions) {
         this.options = options ?? {};
@@ -30,7 +31,29 @@ export class Dataspace implements Partial<Entity> {
         if (is_new) {
             const o = toObserve(v);
             if (o !== void 0) {
-                this.index.addObserver(o.pattern, o.observer);
+                const target = o.observer;
+                const captureMap = new KeyedDictionary<Array<AnyValue>, Handle, Ref>();
+                const observer: IndexObserver = {
+                    onAssert(captures) {
+                        captureMap.set(captures, Turn.active.assert(target, captures));
+                    },
+                    onRetract(vs) {
+                        Turn.active.retract(captureMap.get(vs));
+                        captureMap.delete(vs);
+                    },
+                    onMessage(vs) {
+                        Turn.active.message(target, vs);
+                    },
+                    onRemoval() {
+                        captureMap.forEach((handle, _captures) => Turn.active.retract(handle));
+                    },
+                    dump(): string {
+                        return Array.from(captureMap.entries()).map((handle, values) =>
+                            `captured ${stringify(values)} handle ${handle}`).join('\n');
+                    },
+                };
+                this.observerMap.set(target, observer);
+                this.index.addObserver(o.pattern, observer);
             }
             if (this.options.dumpIndex ?? false) this.index.dump();
         }
@@ -45,7 +68,11 @@ export class Dataspace implements Partial<Entity> {
         if (is_last) {
             const o = toObserve(v);
             if (o !== void 0) {
-                this.index.removeObserver(o.pattern, o.observer);
+                const io = this.observerMap.get(o.observer);
+                if (io !== void 0) {
+                    this.index.removeObserver(o.pattern, io);
+                    this.observerMap.delete(o.observer);
+                }
             }
             if (this.options.dumpIndex ?? false) this.index.dump();
         }

packages/core/src/runtime/skeleton.ts

@@ -1,8 +1,8 @@
 /// SPDX-License-Identifier: GPL-3.0-or-later
 /// SPDX-FileCopyrightText: Copyright © 2016-2023 Tony Garnock-Jones <tonyg@leastfixedpoint.com>
 
-import { Set, Dictionary, KeyedDictionary, IdentityMap, stringify, Value, embed } from '@preserves/core';
+import { Set, Dictionary, KeyedDictionary, IdentitySet, stringify, Value } from '@preserves/core';
-import { AnyValue, Assertion, Handle, Ref, Turn } from './actor.js';
+import { AnyValue, Assertion, Ref } from './actor.js';
 import { Bag, ChangeDescription } from './bag.js';
 import * as Stack from './stack.js';
 import * as P from '../gen/dataspacePatterns.js';
@@ -18,11 +18,19 @@ const _nop = function() {};
 
 const INDENT = '    ';
 
+export interface IndexObserver {
+    onAssert(captures: Assertion[]): void;
+    onRetract(captures: Assertion[]): void;
+    onMessage(captures: Assertion[]): void;
+    onRemoval(): void;
+    dump?(): string;
+}
+
 export class Index {
     readonly allAssertions: Bag<Ref> = new Bag();
     readonly root: Node = new Node(new Continuation(new Set()));
 
-    addObserver(pattern: P.Pattern, observer: Ref) {
+    addObserver(pattern: P.Pattern, observer: IndexObserver) {
         let {constPaths, constValues, capturePaths} = analysePattern(pattern);
         const continuation = this.root.extend(pattern);
         let constValMap = continuation.leafMap.get(constPaths);
@@ -52,13 +60,11 @@ export class Index {
             observerGroup = new ObserverGroup(cachedCaptures);
             leaf.observerGroups.set(capturePaths, observerGroup);
         }
-        const captureMap: KeyedDictionary<Array<AnyValue>, Handle, Ref> = new KeyedDictionary();
+        observerGroup.observers.add(observer);
-        observerGroup.observers.set(observer, captureMap);
+        observerGroup.cachedCaptures.forEach((_count, captures) => observer.onAssert(captures));
-        observerGroup.cachedCaptures.forEach((_count, captures) =>
-            captureMap.set(captures, Turn.active.assert(observer, captures)));
     }
 
-    removeObserver(pattern: P.Pattern, observer: Ref) {
+    removeObserver(pattern: P.Pattern, observer: IndexObserver) {
         let {constPaths, constValues, capturePaths} = analysePattern(pattern);
         const continuation = this.root.extend(pattern);
         let constValMap = continuation.leafMap.get(constPaths);
@@ -67,11 +73,8 @@ export class Index {
         if (!leaf) return;
         let observerGroup = leaf.observerGroups.get(capturePaths);
         if (!observerGroup) return;
-        const captureMap = observerGroup.observers.get(observer);
+        observer.onRemoval();
-        if (captureMap) {
+        observerGroup.observers.delete(observer);
-            captureMap.forEach((handle, _captures) => Turn.active.retract(handle));
-            observerGroup.observers.delete(observer);
-        }
         if (observerGroup.observers.size === 0) {
             leaf.observerGroups.delete(capturePaths);
         }
@@ -92,9 +95,9 @@ export class Index {
                     (c, v) => c.cachedAssertions.add(v),
                     (l, v) => l.cachedAssertions.add(v),
                     (h, vs) => {
-                        if (h.cachedCaptures.change(vs, +1) === ChangeDescription.ABSENT_TO_PRESENT)
+                        if (h.cachedCaptures.change(vs, +1) === ChangeDescription.ABSENT_TO_PRESENT) {
-                            h.observers.forEach((captureMap, observer) =>
+                            h.observers.forEach(observer => observer.onAssert(vs));
-                                captureMap.set(vs, Turn.active.assert(observer, vs)));
+                        }
                     });
                 return true;
 
@@ -105,11 +108,9 @@ export class Index {
                     (c, v) => c.cachedAssertions.delete(v),
                     (l, v) => l.cachedAssertions.delete(v),
                     (h, vs) => {
-                        if (h.cachedCaptures.change(vs, -1) === ChangeDescription.PRESENT_TO_ABSENT)
+                        if (h.cachedCaptures.change(vs, -1) === ChangeDescription.PRESENT_TO_ABSENT) {
-                            h.observers.forEach((captureMap, _observer) => {
+                            h.observers.forEach(observer => observer.onRetract(vs));
-                                Turn.active.retract(captureMap.get(vs));
+                        }
-                                captureMap.delete(vs);
-                            });
                     });
                 return true;
 
@@ -128,7 +129,7 @@ export class Index {
 
     deliverMessage(v: Assertion, leafCallback: (l: Leaf, v: Assertion) => void = _nop) {
         this.root.modify(EventType.MESSAGE, v, _nop, leafCallback, (h, vs) =>
-            h.observers.forEach((_captureMap, observer) => Turn.active.message(observer, vs)));
+            h.observers.forEach(observer => observer.onMessage(vs)));
     }
 
     dump() {
@@ -301,7 +302,7 @@ class Leaf {
 
 class ObserverGroup {
     readonly cachedCaptures: Bag<Ref, Array<AnyValue>>;
-    readonly observers: IdentityMap<Ref, KeyedDictionary<Array<AnyValue>, Handle, Ref>> = new IdentityMap();
+    readonly observers = new IdentitySet<IndexObserver>();
 
     constructor(cachedCaptures: Bag<Ref, Array<AnyValue>>) {
         this.cachedCaptures = cachedCaptures;
@@ -309,11 +310,12 @@ class ObserverGroup {
 
     dump(paths: Path[], indent: string) {
         dumpBag(this.cachedCaptures, indent);
-        this.observers.forEach((valueMap, observer) => {
+        this.observers.forEach(observer => {
-            console.log(indent + `${stringify(embed(observer))} projecting ${stringify(paths)}`);
+            console.log(indent + `${observer} projecting ${stringify(paths)}`);
-            valueMap.forEach((handle, values) => {
+            const d = observer.dump?.();
-                console.log(indent + INDENT + `captured ${stringify(values)} handle ${handle}`);
+            if (d) {
-            });
+                console.log(indent + INDENT + d.split(/\n/).join(indent + INDENT + '\n'));
+            }
         });
     }
 }