fixes to sync observability - WIP

Author: sbazerque

src/mesh/service/Mesh.ts

@@ -18,6 +18,7 @@ import { ObjectSpawnAgent, SpawnCallback } from 'mesh/agents/spawn/ObjectSpawnAg
 import { ObjectInvokeAgent } from 'mesh/agents/spawn/ObjectInvokeAgent';
 import { PeerGroupState } from 'mesh/agents/peer/PeerGroupState';
 import { Resources } from 'spaces/Resources';
+import { MeshInterface } from './remoting/MeshInterface';
 
 
 /* Connect to the Hyper Hyper Space service mesh.
@@ -106,7 +107,7 @@ class CannotInferPeerGroup extends Error {
     }
 }
 
-class Mesh {
+class Mesh implements MeshInterface {
 
     static syncCommandsLog = new Logger('mesh-sync-commands', LogLevel.INFO);
 
@@ -257,7 +258,7 @@ class Mesh {
         return this.registerUsage({type: 'object-sync', objHash: obj.getLastHash(), peerGroupId: peerGroupId}, usageToken);
     }
 
-    syncManyObjectsWithPeerGroup(peerGroupId: string, objs: IterableIterator<HashedObject>, mode:SyncMode=SyncMode.full, usageTokens?: Map<Hash, UsageToken>): Map<Hash, UsageToken>{
+    syncManyObjectsWithPeerGroup(peerGroupId: string, objs: IterableIterator<HashedObject>, mode:SyncMode=SyncMode.full, usageTokens?: Map<Hash, UsageToken>): Map<Hash, UsageToken> {
 
         const tokens = new Map<Hash, UsageToken>();
 
@@ -457,9 +458,6 @@ class Mesh {
         objectInvokeAgent.sendRequest(object, receiver, receiverLinkupServers, senderEndpoint);
     }
 
-
-
-
     getSyncAgentFor(peerGroupId: PeerGroupId, mutHash: Hash): StateSyncAgent|undefined {
         return this.syncAgents.get(peerGroupId)?.get(mutHash);
     }

src/mesh/service/remoting/MeshInterface.ts

@@ -0,0 +1,46 @@
+import { Identity } from 'data/identity';
+import { Hash, HashedObject, MutableObject } from 'data/model';
+import { LinkupAddress } from 'net/linkup';
+import { ObjectDiscoveryReply } from '../../agents/discovery';
+import { SpawnCallback } from '../../agents/spawn';
+
+import { AsyncStream } from 'util/streams';
+import { Endpoint } from '../../agents/network';
+import { PeerGroupAgentConfig, PeerGroupState } from '../../agents/peer';
+import { SyncObserver, SyncState } from '../../agents/state';
+import { PeerGroupInfo, SyncMode, UsageToken } from '../Mesh';
+
+type PeerGroupId = string;
+
+interface MeshInterface {
+    
+    joinPeerGroup(pg: PeerGroupInfo, config?: PeerGroupAgentConfig, usageToken?: UsageToken): UsageToken;
+    leavePeerGroup(token: UsageToken): void;
+    
+    getPeerGroupState(peerGroupId: PeerGroupId): Promise<PeerGroupState|undefined>;
+
+    syncObjectWithPeerGroup(peerGroupId: PeerGroupId, obj: HashedObject, mode?:SyncMode, usageToken?: UsageToken): UsageToken;
+    syncManyObjectsWithPeerGroup(peerGroupId: PeerGroupId, objs: IterableIterator<HashedObject>, mode?:SyncMode, usageTokens?: Map<Hash, UsageToken>): Map<Hash, UsageToken>;
+    stopSyncObjectWithPeerGroup(usageToken: UsageToken): void;
+    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;
+
+    startObjectBroadcast(object: HashedObject, linkupServers: string[], replyEndpoints: Endpoint[], broadcastedSuffixBits?: number, usageToken?: UsageToken): UsageToken;
+    stopObjectBroadcast(token: UsageToken): void;
+
+    findObjectByHash(hash: Hash, linkupServers: string[], replyAddress: LinkupAddress, count?: number, maxAge?: number, strictEndpoints?: boolean, includeErrors?: boolean) : AsyncStream<ObjectDiscoveryReply>;
+    findObjectByHashSuffix(hashSuffix: string, linkupServers: string[], replyAddress: LinkupAddress, count?: number, maxAge?: number, strictEndpoints?: boolean, includeErrors?: boolean) : AsyncStream<ObjectDiscoveryReply>;
+    findObjectByHashRetry(hash: Hash, linkupServers: string[], replyAddress: LinkupAddress, count?: number): void;
+    findObjectByHashSuffixRetry(hashSuffix: string, linkupServers: string[], replyAddress: LinkupAddress, count?: number): void;
+
+    addObjectSpawnCallback(callback: SpawnCallback, receiver: Identity, linkupServers: Array<string>, spawnId?: string): void;
+    sendObjectSpawnRequest(object: HashedObject, sender: Identity, receiver: Identity, senderEndpoint: Endpoint, receiverLinkupServers: Array<string>, spawnId?: string): void;
+
+    shutdown(): void;
+
+}
+
+export { MeshInterface }

src/mesh/service/remoting/MeshProxy.ts

@@ -1,4 +1,4 @@
-import { PeerGroupAgentConfig, PeerInfo, PeerSource } from 'mesh/agents/peer';
+import { ObjectDiscoveryPeerSource, PeerGroupAgentConfig, PeerInfo, PeerSource } from 'mesh/agents/peer';
 import { Mesh, PeerGroupInfo, SyncMode, UsageToken } from 'mesh/service/Mesh';
 import { MeshCommand,
     JoinPeerGroup, LeavePeerGroup,
@@ -14,7 +14,7 @@ import { MeshCommand,
     ForwardPeerGroupState} from './MeshHost';
 
 import { RNGImpl } from 'crypto/random';
-import { Context, Hash, HashedObject } from 'data/model';
+import { Context, Hash, HashedObject, MutableObject } from 'data/model';
 import { AsyncStream, BufferedAsyncStream, BufferingAsyncStreamSource } from 'util/streams';
 import { ObjectDiscoveryReply } from 'mesh/agents/discovery';
 import { Endpoint } from 'mesh/agents/network';
@@ -23,12 +23,15 @@ import { WebRTCConnectionEvent, WebRTCConnectionsHost } from 'net/transport';
 import { Identity } from 'data/identity';
 import { ObjectSpawnAgent, SpawnCallback } from 'mesh/agents/spawn';
 import { PeerGroupState } from 'mesh/agents/peer/PeerGroupState';
+import { Resources } from 'spaces/Resources';
+import { MeshInterface } from './MeshInterface';
+import { SyncState, SyncObserver } from 'mesh/agents/state';
 
 /* Access a mesh remotely, see the MeshHost class. */
 
 type RequestId = string;
 
-class MeshProxy {
+class MeshProxy implements MeshInterface {
 
     commandForwardingFn: (cmd: MeshCommand) => void;
     discoveryStreamSources: Map<string, BufferingAsyncStreamSource<ObjectDiscoveryReply>>;
@@ -497,6 +500,38 @@ class MeshProxy {
 
     }
 
+    // We do not need to bridge this request to the MeshHost: the ObjectDiscoveryPeerSource receives a reference
+    // to this mesh, that's already bridged.
+    async getDiscoveryPeerGroup(obj: HashedObject, resources?: Resources) : Promise<PeerGroupInfo> {
+
+        resources = resources || obj.getResources();
+
+        if (resources === undefined) {
+            throw new Error('Could not find a valid resources object to use for the discovery peer group.');
+        }
+
+        let localPeer = resources.getPeersForDiscovery()[0];
+        let peerSource = new ObjectDiscoveryPeerSource(this as any as Mesh, obj, resources.config.linkupServers, LinkupAddress.fromURL(localPeer.endpoint, localPeer.identity), resources.getEndointParserForDiscovery());
+
+        return {
+            id: Mesh.discoveryPeerGroupId(obj),
+            localPeer: localPeer,
+            peerSource: peerSource
+        };
+
+    }
+
+    getSyncState(_mut: MutableObject, _peerGroupId?: string | undefined): Promise<SyncState | undefined> {
+        throw new Error('Method not implemented.');
+    }
+
+    addSyncObserver(_obs: SyncObserver, _mut: MutableObject, _peerGroupId?: string | undefined): void {
+        throw new Error('Method not implemented.');
+    }
+
+    removeSyncObserver(_obs: SyncObserver, _mut: MutableObject, _peerGroupId?: string | undefined): void {
+        throw new Error('Method not implemented.');
+    }
 
 }
 

src/mesh/service/webworker/WebWorkerMeshProxy.ts

@@ -137,7 +137,7 @@ class WebWorkerMeshProxy {
     }
 
     getMesh(): Mesh {
-        return this.proxy as any as Mesh;
+        return this.proxy as unknown as Mesh;
     }
 
 }