renamed HashedObject's addMutationObserver to addObserver

added new sync state visibility APIs (WIP...)

Author: sbazerque

src/data/model/immutable/HashedObject.ts

@@ -341,11 +341,11 @@ abstract class HashedObject {
         return new EventRelay(this, subObservers);
     }
 
-    addMutationObserver(obs: MutationObserver) {
+    addObserver(obs: MutationObserver) {
         this.getMutationEventSource().addObserver(obs);
     }
 
-    removeMutationObserver(obs: MutationObserver) {
+    removeObserver(obs: MutationObserver) {
         this._mutationEventSource?.removeObserver(obs);
     }
 

src/data/model/mutable/MutableObject.ts

@@ -4,7 +4,7 @@ import { HashedObject } from '../immutable/HashedObject';
 import { Context } from '../literals/Context';
 import { Hash } from '../hashing/Hashing';
 import { Logger, LogLevel } from 'util/logging';
-import { HeaderBasedSyncAgent, StateSyncAgent, StateFilter } from 'mesh/agents/state';
+import { HeaderBasedSyncAgent, StateSyncAgent, StateFilter, SyncState, SyncObserver } from 'mesh/agents/state';
 import { PeerGroupAgent } from 'mesh/agents/peer';
 import { HashedSet } from '../immutable/HashedSet';
 import { HashReference } from '../immutable/HashReference';
@@ -603,7 +603,7 @@ abstract class  MutableObject extends HashedObject {
                 for (const elmt of aliases) {
                     if (!seen.has(elmt)) {
                         seen.add(elmt);
-                        MutableObject.addEventRelayForElmt(own, hash, elmt);
+                        MutableObject.removeEventRelayForElmt(own, hash, elmt);
                     }
                 }
             }    
@@ -615,10 +615,40 @@ abstract class  MutableObject extends HashedObject {
         //return new TerminalOpsSyncAgent(peerGroupAgent, this.getLastHash(), this.getStore(), this._acceptedMutationOpClasses);
     }
 
+    getSyncAgentId(peerGroupId: string) {
+        return HeaderBasedSyncAgent.syncAgentIdFor(this.getLastHash(), peerGroupId);
+    }
+
     getSyncAgentStateFilter() : StateFilter | undefined {
         return undefined;
     }
 
+    async getSyncState(peerGroupId?: string): Promise<SyncState|undefined> {
+        return this.getResources()?.mesh.getSyncState(this, peerGroupId);
+    }
+
+    addSyncObserver(obs: SyncObserver, peerGroupId?: string) {
+
+        const mesh = this.getResources()?.mesh;
+
+        if (mesh === undefined) {
+            throw new Error('Trying to add a sync observer, but object ' + this.hash() + ' does not have a mesh resource.');
+        }
+
+        mesh.addSyncObserver(obs, this, peerGroupId);
+    }
+
+    removeSyncObserver(obs: SyncObserver, peerGroupId?: string) {
+
+        const mesh = this.getResources()?.mesh;
+
+        if (mesh === undefined) {
+            throw new Error('Trying to add a sync observer, but object ' + this.hash() + ' does not have a mesh resource.');
+        }
+
+        mesh.removeSyncObserver(obs, this, peerGroupId);
+    }
+
     getAcceptedMutationOpClasses() : Array<string> {
         return this._acceptedMutationOpClasses;
     }

src/mesh/agents/state.ts

@@ -3,6 +3,7 @@ export * from './state/StateSyncAgent';
 export * from './state/TerminalOpsState';
 export * from './state/TerminalOpsSyncAgent';
 export * from './state/HeaderBasedSyncAgent';
+export * from './state/SyncObserverAgent';
 export * from './state/history/HeaderBasedState';
 export * from './state/history/HistoryProvider';
 export * from './state/history/HistorySynchronizer';

src/mesh/agents/state/HeaderBasedSyncAgent.ts

@@ -6,13 +6,15 @@ import { Endpoint } from '../network/NetworkAgent';
 import { PeerGroupAgent } from '../peer/PeerGroupAgent';
 import { PeeringAgentBase } from '../peer/PeeringAgentBase';
 import { HeaderBasedState } from './history/HeaderBasedState';
-import { AgentStateUpdateEvent, GossipEventTypes } from './StateGossipAgent';
+import { AgentStateUpdateEvent, GossipEventTypes, StateGossipAgent } from './StateGossipAgent';
 import { StateSyncAgent } from './StateSyncAgent';
 
 import { HistorySynchronizer } from './history/HistorySynchronizer';
 import { HistoryProvider, MessageType, SyncMsg } from './history/HistoryProvider';
 import { OpHeader, OpHeaderLiteral } from 'data/history/OpHeader';
 import { Resources } from 'spaces/Resources';
+import { EventRelay } from 'util/events';
+import { SyncObserverEventTypes, SyncState, SyncStateUpdateEvent } from './SyncObserverAgent';
 
 /*
  *  Important notice: The constructor of the HeaderBasedSyncAgent can receive either an instance or just the
@@ -36,7 +38,7 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
 
     static MaxRequestsPerRemote = 2;
 
-    mutableObj?: MutableObject;
+    mutableObj: MutableObject;
     mutableObjHash: Hash;
     acceptedMutationOpClasses: string[];
     stateOpFilter?: StateFilter;
@@ -55,20 +57,19 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
     synchronizer : HistorySynchronizer;
     provider     : HistoryProvider;
 
+    syncEventSource?: EventRelay<HashedObject>;
+
     terminated = false;
 
     controlLog: Logger;
     messageLog: Logger;
 
-    constructor(peerGroupAgent: PeerGroupAgent, mutableObjOrHash: MutableObject|Hash, resources: Resources, acceptedMutationOpClasses : string[], stateOpFilter?: StateFilter) {
+    constructor(peerGroupAgent: PeerGroupAgent, mutableObj: MutableObject, resources: Resources, acceptedMutationOpClasses : string[], stateOpFilter?: StateFilter) {
         super(peerGroupAgent);
 
-        if (mutableObjOrHash instanceof MutableObject) {
-            this.mutableObj  = mutableObjOrHash;
-            this.mutableObjHash = mutableObjOrHash.hash();    
-        } else {
-            this.mutableObjHash =  mutableObjOrHash;
-        }
+        this.mutableObj = mutableObj;
+        this.mutableObjHash = mutableObj.hash();
+
         this.acceptedMutationOpClasses = acceptedMutationOpClasses;
         this.stateOpFilter = stateOpFilter;
 
@@ -86,16 +87,14 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
         this.messageLog = HeaderBasedSyncAgent.messageLog;
     }
 
-
-
-    
     getAgentId(): string {
         return HeaderBasedSyncAgent.syncAgentIdFor(this.mutableObjHash, this.peerGroupAgent.peerGroupId);
     }
 
     ready(pod: AgentPod): void {
 
         this.pod = pod;
+
         this.updateStateFromStore().then(async () => {
                 if (this.stateOpHeadersByOpHash !== undefined) {
                     for (const opHistory of this.stateOpHeadersByOpHash.values()) {
@@ -112,6 +111,14 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
 
     shutdown(): void {
 
+        const ev: SyncStateUpdateEvent = {
+            emitter: this.mutableObj,
+            action: SyncObserverEventTypes.SyncStateUpdate,
+            data: {inSync: false, synchronizing: false, opsToFetch: 0, remoteStateHashes: {}}
+        };
+
+        this.syncEventSource?.emit(ev);
+
         this.terminated = true;
         this.synchronizer.shutdown();
 
@@ -206,6 +213,16 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
             await this.synchronizer.onNewLocalOp(op);
             await this.updateStateFromStore();  
         }
+
+        if (this.syncEventSource !== undefined) {
+            const ev: SyncStateUpdateEvent = {
+                emitter: this.mutableObj,
+                action: SyncObserverEventTypes.SyncStateUpdate,
+                data: this.getSyncState()
+            };
+
+            this.syncEventSource.emit(ev);
+        }
     };
 
     literalIsValidOp(literal?: Literal, log=false): boolean { 
@@ -324,6 +341,46 @@ class HeaderBasedSyncAgent extends PeeringAgentBase implements StateSyncAgent {
         return this.synchronizer.expectingMoreOps(receivedOpHashes);
     }
 
+    getSyncEventSource(): EventRelay<HashedObject> {
+
+        if (this.syncEventSource === undefined) {
+            this.syncEventSource = this.createSyncEventSource();
+        }
+
+        return this.syncEventSource;
+    }
+
+    createSyncEventSource(): EventRelay<HashedObject> {
+        return new EventRelay(this.mutableObj as HashedObject, new Map());
+    }
+
+    getSyncState(): SyncState {
+
+        const gossipAgent = this.pod?.getAgent(StateGossipAgent.agentIdForGossipId(this.getAgentId())) as StateGossipAgent;
+
+        const remoteStateHashes = gossipAgent.getAllRemoteStateForAgent(this.getAgentId());
+        const localStateHash    = gossipAgent.getLocalStateForAgent(this.getAgentId());
+
+        let inSync = true;
+        for (const remoteHash of Object.values(remoteStateHashes)) {
+            if (remoteHash !== localStateHash) {
+                inSync = false;
+                break;
+            }
+        }
+
+        const opsToFetch = this.synchronizer.discoveredHistory.contents.size;
+
+        return {remoteStateHashes: remoteStateHashes, localStateHash: localStateHash, inSync: inSync, opsToFetch: opsToFetch, synchronizing: true}
+    }
+
+    getMutableObject(): MutableObject {
+        return this.mutableObj;
+    }
+
+    getPeerGroupId(): string {
+        return this.peerGroupAgent.peerGroupId;
+    }
 }
 
 export { SyncMsg as HistoryMsg, HeaderBasedSyncAgent, StateFilter }

src/mesh/agents/state/StateGossipAgent.ts

@@ -70,7 +70,7 @@ enum GossipType {
 
 interface SendFullState { 
     type: GossipType.SendFullState,
-    state: {entries: [AgentId, Hash][], hashes: Hash[]} //HashedMap<AgentId, Hash>.toArrays
+    state: {entries: [AgentId, Hash][], hashes: Hash[]}
 };
 
 interface SendStateObject {
@@ -121,7 +121,7 @@ enum SendingReason {
 
 class StateGossipAgent extends PeeringAgentBase {
 
-    static agentIdForGossip(gossipId: string) {
+    static agentIdForGossipId(gossipId: string) {
         return 'state-gossip-agent-for-' + gossipId;
     }
 
@@ -164,9 +164,9 @@ class StateGossipAgent extends PeeringAgentBase {
 
 
 
-    constructor(topic: string, peerNetwork: PeerGroupAgent) {
+    constructor(gossipId: string, peerNetwork: PeerGroupAgent) {
         super(peerNetwork);
-        this.gossipId = topic;
+        this.gossipId = gossipId;
 
         this.trackedAgentIds = new Set();
         this.localState  = new Map();
@@ -211,7 +211,7 @@ class StateGossipAgent extends PeeringAgentBase {
     }
 
     getAgentId(): string {
-        return StateGossipAgent.agentIdForGossip(this.gossipId);
+        return StateGossipAgent.agentIdForGossipId(this.gossipId);
     }
 
     getNetwork() : AgentPod {
@@ -471,7 +471,12 @@ class StateGossipAgent extends PeeringAgentBase {
 
         if (gossip.type === GossipType.RequestStateObject) {
             this.controlLog.trace('Recevied state request for ' + gossip.agentId);
-            this.sendStateObject(source, gossip.agentId, SendingReason.Request);    
+            if (this.localStateObjects.get(gossip.agentId)) {
+                this.sendStateObject(source, gossip.agentId, SendingReason.Request);    
+            } else {
+                this.controlLog.debug('Recevied state request for ' + gossip.agentId + ', but no state was found: not sending')
+            }
+            
         }
     }
 
@@ -557,7 +562,7 @@ class StateGossipAgent extends PeeringAgentBase {
 
             }
 
-            if (receivedOldState && this.localState.get(agentId) !== state) {
+            if (receivedOldState && this.localState.get(agentId) !== state && this.localStateObjects.get(agentId) !== undefined) {
                 this.peerMessageLog.trace('Received old state for ' + agentId + ' from ' + sender + ', sending our own state over there.');
                 this.sendStateObject(sender, agentId);
             }
@@ -680,6 +685,26 @@ class StateGossipAgent extends PeeringAgentBase {
 
     }
 
+    // for other agents
+
+    getAllRemoteStateForAgent(agentId: AgentId): {[key: Endpoint]: Hash} {
+
+        let state: {[key: Endpoint]: Hash} = {};
+
+        for (const [endpoint, peerState] of this.remoteState.entries()) {
+            const stateHash = peerState.get(agentId);
+            
+            if (stateHash !== undefined) {
+                state[endpoint] = stateHash;
+            }
+        }
+
+        return state;
+    }
+
+    getLocalStateForAgent(agentId: AgentId): Hash|undefined {
+        return this.localState.get(agentId);
+    }
 
 }
 

src/mesh/agents/state/StateSyncAgent.ts

@@ -1,12 +1,19 @@
-import { Hash, HashedObject } from 'data/model';
+import { Hash, HashedObject, MutableObject } from 'data/model';
+import { EventRelay } from 'util/events';
 import { Agent } from '../../service/Agent';
 import { Endpoint } from '../network/NetworkAgent';
+import { SyncState } from './SyncObserverAgent';
 
 interface StateSyncAgent extends Agent {
 
     receiveRemoteState(sender: Endpoint, stateHash: Hash, state: HashedObject) : Promise<boolean>;
     expectingMoreOps(receivedOpHashes?: Set<Hash>): boolean;
 
+    getMutableObject(): MutableObject;
+    getPeerGroupId(): string;
+
+    getSyncState(): SyncState;
+    getSyncEventSource(): EventRelay<HashedObject>;
 }
 
 export { StateSyncAgent }