bridged sync observing to/from WebWorker-based mesh.

sbazerque · sbazerque · commit 19647e54182e · 2022-11-04T11:32:11.000-03:00
diff --git a/src/mesh/agents/state/HeaderBasedSyncAgent.ts b/src/mesh/agents/state/HeaderBasedSyncAgent.ts
@@ -362,11 +362,8 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
                 action: SyncObserverEventTypes.SyncStateUpdate,
                 data: this.getSyncState()
             };
-
-            console.log('new sync state event:', ev);
+            
             this.syncEventSource.emit(ev);
-        } else {
-            console.log('not emitting new sync state event: no relay')
         }
     }
 
diff --git a/src/mesh/agents/state/StateGossipAgent.ts b/src/mesh/agents/state/StateGossipAgent.ts
@@ -696,11 +696,9 @@ class StateGossipAgent extends PeeringAgentBase {
             
             if (stateHash !== undefined) {
                 state[endpoint] = stateHash;
-                console.log('ep=', endpoint, ', hash=', stateHash);
             }
         }
 
-        console.log(state);
         return state;
     }
 
diff --git a/src/mesh/agents/state/SyncObserverAgent.ts b/src/mesh/agents/state/SyncObserverAgent.ts
@@ -61,8 +61,6 @@ class SyncObserverAgent implements Agent {
 
     addSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId: PeerGroupId) {
 
-        console.log('adding sync observer ', obs, ' for ', mut.getLastHash(), ' in peer group ', peerGroupId)
-
         if (this.pod === undefined) {
             throw new Error('Trying to add a sync observer, but the SyncObserverAgent is not ready.');
         }
@@ -139,12 +137,10 @@ class SyncObserverAgent implements Agent {
                     const syncAgent = this.pod?.getAgent(syncAgentId) as StateSyncAgent;
                     const [_relay, syncAgentObserver] = pair;
                     syncAgent.getSyncEventSource().addObserver(syncAgentObserver);
-                    console.log('+++ adding agent relay ', syncAgent.getSyncEventSource(), ' to ', _relay)
                 } else if (agentEv.content.change === AgentSetChange.Removal) {
                     const syncAgent = this.pod?.getAgent(syncAgentId) as StateSyncAgent|undefined;
                     const [_relay, syncAgentObserver] = pair;
                     syncAgent?.getSyncEventSource().addObserver(syncAgentObserver);
-                    console.log('+++ removing agent relay from ', _relay)
                 }
     
             }
diff --git a/src/mesh/service/Mesh.ts b/src/mesh/service/Mesh.ts
@@ -328,18 +328,16 @@ class Mesh implements MeshInterface {
         }
     }
 
-    addSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId) {
+    async addSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId) {
 
         if (peerGroupId === undefined) {
             peerGroupId = this.inferPeerGroupId(mut);
         }
 
-        console.log('mesh: observer to ', mut.getLastHash(), ' peerGroupId=', peerGroupId);
-
         this.syncObserver.addSyncObserver(obs, mut, peerGroupId);
     }
 
-    removeSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId) {
+    async removeSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId) {
 
         if (peerGroupId === undefined) {
             peerGroupId = this.inferPeerGroupId(mut);
@@ -875,7 +873,6 @@ class Mesh implements MeshInterface {
         if (agents.size !== 1) {
             throw new CannotInferPeerGroup(mutHash);
         } else {
-            console.log('agents ', agents)
             return agents.values().next().value as string;
         }
     }
diff --git a/src/mesh/service/remoting/MeshHost.ts b/src/mesh/service/remoting/MeshHost.ts
@@ -1,10 +1,3 @@
-import { PeerGroupAgentConfig, PeerInfo, PeerSource } from '../../agents/peer';
-import { CannotInferPeerGroup, Mesh, SyncMode, UsageToken } from '../../service/Mesh';
-import { SpawnCallback } from '../../agents/spawn';
-import { PeerGroupState } from '../../agents/peer/PeerGroupState';
-import { ObjectDiscoveryReply } from '../../agents/discovery';
-import { Endpoint } from '../../agents/network';
-
 import { Identity, RSAKeyPair } from 'data/identity';
 import { Context, HashedObject, LiteralContext, MutableObject } from 'data/model';
 import { Hash } from 'data/model';
@@ -15,7 +8,14 @@ import { RNGImpl } from 'crypto/random';
 
 import { Store } from 'storage/store';
 import { LinkupAddress } from 'net/linkup';
-import { SyncState } from 'mesh/agents/state';
+
+import { PeerGroupAgentConfig, PeerInfo, PeerSource } from '../../agents/peer';
+import { CannotInferPeerGroup, Mesh, SyncMode, UsageToken } from '../../service/Mesh';
+import { SpawnCallback } from '../../agents/spawn';
+import { PeerGroupState } from '../../agents/peer/PeerGroupState';
+import { ObjectDiscoveryReply } from '../../agents/discovery';
+import { Endpoint } from '../../agents/network';
+import { SyncEvent, SyncObserver, SyncState } from '../../agents/state';
 
 /* Run a mesh remotely, and access it through a MeshProxy */
 
@@ -31,13 +31,15 @@ import { SyncState } from 'mesh/agents/state';
  * 
  * Ah, the things we do for you, Hyper Hyper Space. */
 
+type ObserverId = string;
+
 type MeshCommand = JoinPeerGroup | LeavePeerGroup | ForwardPeerGroupState |
                    SyncObjectsWithPeerGroup | StopSyncObjectsWithPeerGroup |
                    StartObjectBroadcast | StopObjectBroadcast |
                    FindObjectByHash | FindObjectByHashSuffix |
                    ForwardGetPeersReply | ForwardGetPeerForEndpointReply |
                    AddObjectSpawnCallback | SendObjectSpawnRequest |
-                   ForwardSyncState |
+                   ForwardSyncState | AddSyncObserver | RemoveSyncObserver |
                    Shutdown;
 
 type JoinPeerGroup = {
@@ -147,6 +149,20 @@ type ForwardSyncState = {
     peerGroupId?: string
 }
 
+type AddSyncObserver = {
+    type: 'add-sync-observer',
+    observerId: ObserverId,
+    mutLiteralContext: LiteralContext,
+    peerGroupId?: string
+}
+
+type RemoveSyncObserver = {
+    type: 'remove-sync-observer',
+    observerId: ObserverId,
+    mutLiteralContext: LiteralContext,
+    peerGroupId?: string
+}
+
 type Shutdown = {
     type: 'shutdown'
 }
@@ -167,7 +183,8 @@ type ForwardGetPeerForEndpointReply = {
 
 type PeerInfoContext = { endpoint: Endpoint, identityHash: Hash, identity?: LiteralContext };
 
-type CommandStreamedReply = LiteralObjectDiscoveryReply | DiscoveryEndReply | ObjectSpawnCallback | PeerGroupStateReply | SyncStateReply;
+type CommandStreamedReply = LiteralObjectDiscoveryReply | DiscoveryEndReply | ObjectSpawnCallback | PeerGroupStateReply | 
+                            SyncStateReply | AddSyncObserverReply | RemoveSyncObserverReply | SyncObserverEventReply;
 
 type LiteralObjectDiscoveryReply = {
     type: 'object-discovery-reply'
@@ -210,6 +227,27 @@ type SyncStateReply = {
     errorType?: 'infer-peer-group'
 }
 
+type AddSyncObserverReply = {
+    type: 'add-sync-observer-reply',
+    observerId: ObserverId,
+    errorType?: 'infer-peer-group',
+    error?: string
+}
+
+type RemoveSyncObserverReply = {
+    type: 'remove-sync-observer-reply',
+    observerId: ObserverId,
+    errorType?: 'infer-peer-group',
+    error?: string
+}
+
+type SyncObserverEventReply = {
+    type: 'sync-observer-event-reply',
+    observerId: ObserverId,
+    action: string,
+    data: SyncState
+}
+
 type PeerSourceRequest = GetPeersRequest | GetPeerForEndpointRequest;
 
 type GetPeersRequest = { 
@@ -249,17 +287,23 @@ class MeshHost {
             type === 'forward-get-peer-for-endpoint-reply' || 
             type === 'add-object-spawn-callback' ||
             type === 'send-object-spawn-callback' ||
-            type === 'forward-sync-state'
+            type === 'forward-sync-state' ||
+            type === 'add-sync-observer' ||
+            type === 'remove-sync-observer'
             );
     }
 
     static isStreamedReply(msg: any): boolean {
         const type = msg?.type;
 
         return (type === 'object-discovery-reply' ||
-                type === 'object-discovery-end'   || 
-                type === 'object-spawn-callback'  ||
-                type === 'peer-group-state-reply');
+                type === 'object-discovery-end' || 
+                type === 'object-spawn-callback' ||
+                type === 'peer-group-state-reply' ||
+                type === 'sync-state-reply' ||
+                type === 'add-sync-observer-reply' ||
+                type === 'remove-sync-observer-reply' ||
+                type === 'sync-observer-event-reply');
     }
 
     static isPeerSourceRequest(msg: any): boolean {
@@ -280,6 +324,8 @@ class MeshHost {
 
     stores: Map<string, Map<string, Store>>;
 
+    syncObservers: Map<ObserverId, SyncObserver>;
+
     constructor(mesh: Mesh, streamedReplyCb: (resp: CommandStreamedReply) => void, peerSourceReqCb: (req: PeerSourceRequest) => void) {
         this.mesh = mesh;
         this.streamedReplyCb = streamedReplyCb;
@@ -288,6 +334,7 @@ class MeshHost {
         this.pendingPeerForEndpointRequests = new Map();
         this.spawnCallbacks = new Map();
         this.stores = new Map();
+        this.syncObservers = new Map();
     }
 
     execute(command: MeshCommand) : void {
@@ -371,11 +418,72 @@ class MeshHost {
                             reply.errorType = 'infer-peer-group';
                         }
 
-                        reply.error = reason;
+                        reply.error = reason.toString();
 
                         this.streamedReplyCb(reply);
                     });
-                    
+
+        } else if (command.type === 'add-sync-observer') {
+            let obs: SyncObserver = (ev: SyncEvent) => {
+
+                const reply: SyncObserverEventReply = {
+                    type: 'sync-observer-event-reply',
+                    observerId: command.observerId,
+                    action: ev.action,
+                    data: ev.data
+                }
+
+                this.streamedReplyCb(reply);
+            }
+
+            const reply: AddSyncObserverReply = {
+                type: 'add-sync-observer-reply',
+                observerId: command.observerId
+            }
+
+            try {
+                let mut = HashedObject.fromLiteralContext(command.mutLiteralContext) as MutableObject;
+                this.mesh.addSyncObserver(obs, mut, command.peerGroupId);
+            } catch (e: any) {
+                if (e instanceof CannotInferPeerGroup) {
+                    reply.errorType = 'infer-peer-group'
+                }
+                reply.error = e.toString();
+            }
+
+            if (reply.error === undefined && reply.errorType === undefined) {
+                this.syncObservers.set(command.observerId, obs);
+            }
+            
+            this.streamedReplyCb(reply);
+
+        } else if (command.type === 'remove-sync-observer') {
+
+            const reply: RemoveSyncObserverReply = {
+                type: 'remove-sync-observer-reply',
+                observerId: command.observerId
+            }
+
+            try {
+                const obs = this.syncObservers.get(command.observerId);
+
+                if (obs !== undefined) {
+                    let mut = HashedObject.fromLiteralContext(command.mutLiteralContext) as MutableObject;
+                    this.mesh.removeSyncObserver(obs, mut, command.peerGroupId);
+                }
+            } catch (e: any) {
+                if (e instanceof CannotInferPeerGroup) {
+                    reply.errorType = 'infer-peer-group'
+                }
+                reply.error = e.toString();
+            }
+
+            if (reply.error === undefined && reply.errorType === undefined) {
+                this.syncObservers.delete(command.observerId);
+            }
+
+            this.streamedReplyCb(reply);
+
         } else if (command.type === 'sync-objects-with-peer-group') {
             const syncObjs = command as SyncObjectsWithPeerGroup;
 
@@ -675,7 +783,7 @@ export { MeshHost, MeshCommand,
          SyncObjectsWithPeerGroup, StopSyncObjectsWithPeerGroup,
          StartObjectBroadcast, StopObjectBroadcast,
          FindObjectByHash, FindObjectByHashSuffix, AddObjectSpawnCallback, SendObjectSpawnRequest,
-         ForwardSyncState,
+         ForwardSyncState, AddSyncObserver, RemoveSyncObserver,
          Shutdown,
          CommandStreamedReply, LiteralObjectDiscoveryReply, DiscoveryEndReply, ObjectSpawnCallback, PeerGroupStateReply, LiteralPeerInfo, 
          ForwardGetPeersReply, ForwardGetPeerForEndpointReply,
diff --git a/src/mesh/service/remoting/MeshInterface.ts b/src/mesh/service/remoting/MeshInterface.ts
@@ -25,8 +25,8 @@ interface MeshInterface {
     stopSyncManyObjectsWithPeerGroup(tokens: IterableIterator<UsageToken>): void;
 
     getSyncState(mut: MutableObject, peerGroupId?: string): Promise<SyncState|undefined>;
-    addSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId): void;
-    removeSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId): void;
+    addSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId): Promise<void>;
+    removeSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId): Promise<void>;
 
     startObjectBroadcast(object: HashedObject, linkupServers: string[], replyEndpoints: Endpoint[], broadcastedSuffixBits?: number, usageToken?: UsageToken): UsageToken;
     stopObjectBroadcast(token: UsageToken): void;
diff --git a/src/mesh/service/remoting/MeshProxy.ts b/src/mesh/service/remoting/MeshProxy.ts
diff --git a/test/mesh/mock/LinearStateAgent.ts b/test/mesh/mock/LinearStateAgent.ts

Original file line number	Diff line number	Diff line change
`@@ -362,11 +362,8 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {`
`362`	`362`	`action: SyncObserverEventTypes.SyncStateUpdate,`
`363`	`363`	`data: this.getSyncState()`
`364`	`364`	`};`
`365`		`-`
`366`		`- console.log('new sync state event:', ev);`
	`365`	`+`
`367`	`366`	`this.syncEventSource.emit(ev);`
`368`		`- } else {`
`369`		`- console.log('not emitting new sync state event: no relay')`
`370`	`367`	`}`
`371`	`368`	`}`
`372`	`369`
Original file line number	Diff line number	Diff line change
`@@ -696,11 +696,9 @@ class StateGossipAgent extends PeeringAgentBase {`
`696`	`696`
`697`	`697`	`if (stateHash !== undefined) {`
`698`	`698`	`state[endpoint] = stateHash;`
`699`		`- console.log('ep=', endpoint, ', hash=', stateHash);`
`700`	`699`	`}`
`701`	`700`	`}`
`702`	`701`
`703`		`- console.log(state);`
`704`	`702`	`return state;`
`705`	`703`	`}`
`706`	`704`
Original file line number	Diff line number	Diff line change
`@@ -61,8 +61,6 @@ class SyncObserverAgent implements Agent {`
`61`	`61`
`62`	`62`	`addSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId: PeerGroupId) {`
`63`	`63`
`64`		`- console.log('adding sync observer ', obs, ' for ', mut.getLastHash(), ' in peer group ', peerGroupId)`
`65`		`-`
`66`	`64`	`if (this.pod === undefined) {`
`67`	`65`	`throw new Error('Trying to add a sync observer, but the SyncObserverAgent is not ready.');`
`68`	`66`	`}`
`@@ -139,12 +137,10 @@ class SyncObserverAgent implements Agent {`
`139`	`137`	`const syncAgent = this.pod?.getAgent(syncAgentId) as StateSyncAgent;`
`140`	`138`	`const [_relay, syncAgentObserver] = pair;`
`141`	`139`	`syncAgent.getSyncEventSource().addObserver(syncAgentObserver);`
`142`		`- console.log('+++ adding agent relay ', syncAgent.getSyncEventSource(), ' to ', _relay)`
`143`	`140`	`} else if (agentEv.content.change === AgentSetChange.Removal) {`
`144`	`141`	`const syncAgent = this.pod?.getAgent(syncAgentId) as StateSyncAgent\|undefined;`
`145`	`142`	`const [_relay, syncAgentObserver] = pair;`
`146`	`143`	`syncAgent?.getSyncEventSource().addObserver(syncAgentObserver);`
`147`		`- console.log('+++ removing agent relay from ', _relay)`
`148`	`144`	`}`
`149`	`145`
`150`	`146`	`}`
Original file line number	Diff line number	Diff line change
`@@ -328,18 +328,16 @@ class Mesh implements MeshInterface {`
`328`	`328`	`}`
`329`	`329`	`}`
`330`	`330`
`331`		`- addSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId) {`
	`331`	`+ async addSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId) {`
`332`	`332`
`333`	`333`	`if (peerGroupId === undefined) {`
`334`	`334`	`peerGroupId = this.inferPeerGroupId(mut);`
`335`	`335`	`}`
`336`	`336`
`337`		`- console.log('mesh: observer to ', mut.getLastHash(), ' peerGroupId=', peerGroupId);`
`338`		`-`
`339`	`337`	`this.syncObserver.addSyncObserver(obs, mut, peerGroupId);`
`340`	`338`	`}`
`341`	`339`
`342`		`- removeSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId) {`
	`340`	`+ async removeSyncObserver(obs: SyncObserver, mut: MutableObject, peerGroupId?: PeerGroupId) {`
`343`	`341`
`344`	`342`	`if (peerGroupId === undefined) {`
`345`	`343`	`peerGroupId = this.inferPeerGroupId(mut);`
`@@ -875,7 +873,6 @@ class Mesh implements MeshInterface {`
`875`	`873`	`if (agents.size !== 1) {`
`876`	`874`	`throw new CannotInferPeerGroup(mutHash);`
`877`	`875`	`} else {`
`878`		`- console.log('agents ', agents)`
`879`	`876`	`return agents.values().next().value as string;`
`880`	`877`	`}`
`881`	`878`	`}`