bridged getSyncSate to the web worker! WIP

Author: sbazerque

src/mesh/service/Mesh.ts

@@ -882,4 +882,4 @@ class Mesh implements MeshInterface {
 
 }
 
-export { Mesh, PeerGroupInfo, SyncMode, UsageToken }
+export { Mesh, PeerGroupInfo, SyncMode, UsageToken, CannotInferPeerGroup }

src/mesh/service/remoting/MeshHost.ts

@@ -1,12 +1,12 @@
 import { PeerGroupAgentConfig, PeerInfo, PeerSource } from '../../agents/peer';
-import { Mesh, SyncMode, UsageToken } from '../../service/Mesh';
+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 } from 'data/model';
+import { Context, HashedObject, LiteralContext, MutableObject } from 'data/model';
 import { Hash } from 'data/model';
 
 import { AsyncStream } from 'util/streams';
@@ -15,6 +15,7 @@ import { RNGImpl } from 'crypto/random';
 
 import { Store } from 'storage/store';
 import { LinkupAddress } from 'net/linkup';
+import { SyncState } from 'mesh/agents/state';
 
 /* Run a mesh remotely, and access it through a MeshProxy */
 
@@ -33,9 +34,11 @@ import { LinkupAddress } from 'net/linkup';
 type MeshCommand = JoinPeerGroup | LeavePeerGroup | ForwardPeerGroupState |
                    SyncObjectsWithPeerGroup | StopSyncObjectsWithPeerGroup |
                    StartObjectBroadcast | StopObjectBroadcast |
-                   FindObjectByHash | FindObjectByHashSuffix |  Shutdown |
+                   FindObjectByHash | FindObjectByHashSuffix |
                    ForwardGetPeersReply | ForwardGetPeerForEndpointReply |
-                   AddObjectSpawnCallback | SendObjectSpawnRequest;
+                   AddObjectSpawnCallback | SendObjectSpawnRequest |
+                   ForwardSyncState |
+                   Shutdown;
 
 type JoinPeerGroup = {
     type: 'join-peer-group';
@@ -137,6 +140,13 @@ type SendObjectSpawnRequest = {
     spawnId: string
 }
 
+type ForwardSyncState = {
+    type: 'forward-sync-state',
+    requestId: string,
+    mutLiteralContext: LiteralContext,
+    peerGroupId?: string
+}
+
 type Shutdown = {
     type: 'shutdown'
 }
@@ -157,7 +167,7 @@ type ForwardGetPeerForEndpointReply = {
 
 type PeerInfoContext = { endpoint: Endpoint, identityHash: Hash, identity?: LiteralContext };
 
-type CommandStreamedReply = LiteralObjectDiscoveryReply | DiscoveryEndReply | ObjectSpawnCallback | PeerGroupStateReply;
+type CommandStreamedReply = LiteralObjectDiscoveryReply | DiscoveryEndReply | ObjectSpawnCallback | PeerGroupStateReply | SyncStateReply;
 
 type LiteralObjectDiscoveryReply = {
     type: 'object-discovery-reply'
@@ -192,6 +202,13 @@ type PeerGroupStateReply = {
     remote?: Array<LiteralPeerInfo>
 }
 
+type SyncStateReply = {
+    type: 'sync-state-reply',
+    requestId: string,
+    state?: SyncState
+    error?: string,
+    errorType?: 'infer-peer-group'
+}
 
 type PeerSourceRequest = GetPeersRequest | GetPeerForEndpointRequest;
 
@@ -231,7 +248,8 @@ class MeshHost {
             type === 'forward-get-peers-reply' ||
             type === 'forward-get-peer-for-endpoint-reply' || 
             type === 'add-object-spawn-callback' ||
-            type === 'send-object-spawn-callback'
+            type === 'send-object-spawn-callback' ||
+            type === 'forward-sync-state'
             );
     }
 
@@ -330,8 +348,34 @@ class MeshHost {
 
                 this.streamedReplyCb(reply);
 
-            })
+            });
+
+        } else if (command.type === 'forward-sync-state') {
+
+            const mut = HashedObject.fromLiteralContext(command.mutLiteralContext) as MutableObject;
+            
+            const reply: SyncStateReply = {
+                type: 'sync-state-reply',
+                requestId: command.requestId,
+            }
+
+            this.mesh.getSyncState(mut, command.peerGroupId)
+                     .then((state: SyncState|undefined) => {
+
+                        reply.state = state;
+
+                        this.streamedReplyCb(reply);
+
+                    }).catch((reason: any) => {
+                        if (reason instanceof CannotInferPeerGroup) {
+                            reply.errorType = 'infer-peer-group';
+                        }
+
+                        reply.error = reason;
 
+                        this.streamedReplyCb(reply);
+                    });
+                    
         } else if (command.type === 'sync-objects-with-peer-group') {
             const syncObjs = command as SyncObjectsWithPeerGroup;
 
@@ -631,6 +675,7 @@ export { MeshHost, MeshCommand,
          SyncObjectsWithPeerGroup, StopSyncObjectsWithPeerGroup,
          StartObjectBroadcast, StopObjectBroadcast,
          FindObjectByHash, FindObjectByHashSuffix, AddObjectSpawnCallback, SendObjectSpawnRequest,
+         ForwardSyncState,
          Shutdown,
          CommandStreamedReply, LiteralObjectDiscoveryReply, DiscoveryEndReply, ObjectSpawnCallback, PeerGroupStateReply, LiteralPeerInfo, 
          ForwardGetPeersReply, ForwardGetPeerForEndpointReply,

src/mesh/service/remoting/MeshProxy.ts

@@ -1,5 +1,5 @@
 import { ObjectDiscoveryPeerSource, PeerGroupAgentConfig, PeerInfo, PeerSource } from 'mesh/agents/peer';
-import { Mesh, PeerGroupInfo, SyncMode, UsageToken } from 'mesh/service/Mesh';
+import { CannotInferPeerGroup, Mesh, PeerGroupInfo, SyncMode, UsageToken } from 'mesh/service/Mesh';
 import { MeshCommand,
     JoinPeerGroup, LeavePeerGroup,
     SyncObjectsWithPeerGroup, StopSyncObjectsWithPeerGroup,
@@ -11,7 +11,8 @@ import { MeshCommand,
     PeerInfoContext,
     AddObjectSpawnCallback,
     SendObjectSpawnRequest,
-    ForwardPeerGroupState} from './MeshHost';
+    ForwardPeerGroupState,
+    ForwardSyncState} from './MeshHost';
 
 import { RNGImpl } from 'crypto/random';
 import { Context, Hash, HashedObject, MutableObject } from 'data/model';
@@ -45,7 +46,7 @@ class MeshProxy implements MeshInterface {
     peerSourceRequestIngestFn: (req: PeerSourceRequest) => void;
 
     pendingPeerGroupStates: Map<RequestId, {resolve: (result: PeerGroupState|undefined) => void, reject: (reason: any) => void, timeout: any}>;
-
+    pendingSyncStates: Map<RequestId, {resolve: (result: SyncState|undefined) => void, reject: (reason: any) => void, mut: MutableObject, timeout: any}>;
 
 
     constructor(meshCommandFwdFn: (cmd: MeshCommand) => void, linkupCommandFwdFn?: (cmd: LinkupManagerCommand) => void, webRTCConnEventIngestFn?: (ev: WebRTCConnectionEvent) => void) {
@@ -128,6 +129,33 @@ class MeshProxy implements MeshInterface {
                     }
 
                 }
+            } else if (reply.type === 'sync-state-reply') {
+                const pending = this.pendingSyncStates.get(reply.requestId);
+                const cb = pending?.resolve;
+                const err = pending?.reject;
+                const to =  pending?.timeout;
+                const mut = pending?.mut;
+
+                this.pendingSyncStates.delete(reply.requestId);
+
+                if (to !== undefined) {
+                    window.clearTimeout(to);
+                }
+
+                if (reply.state !== undefined) {
+                
+                    if (cb !== undefined) {        
+                        cb(reply.state);    
+                    }
+                } else {
+                    if (err !== undefined) {
+                        if (reply.errorType === 'infer-peer-group' && mut !== undefined) {
+                            err(new CannotInferPeerGroup(mut.getLastHash()));
+                        } else {
+                            err(reply.error);
+                        }
+                    }
+                }
             }
         }
 
@@ -195,6 +223,7 @@ class MeshProxy implements MeshInterface {
         };
 
         this.pendingPeerGroupStates = new Map();
+        this.pendingSyncStates      = new Map();
     }
 
     getCommandStreamedReplyIngestFn() {
@@ -521,8 +550,32 @@ class MeshProxy implements MeshInterface {
 
     }
 
-    getSyncState(_mut: MutableObject, _peerGroupId?: string | undefined): Promise<SyncState | undefined> {
-        throw new Error('Method not implemented.');
+    getSyncState(mut: MutableObject, peerGroupId?: string | undefined, timeout=10000): Promise<SyncState | undefined> {
+        const p = new Promise<SyncState|undefined>((resolve: (result: SyncState|undefined) => void, reject: (reason: any) => void) => {
+            const requestId = new RNGImpl().randomHexString(128);
+
+            const cmd: ForwardSyncState = {
+                type: 'forward-sync-state',
+                requestId: requestId,
+                peerGroupId: peerGroupId,
+                mutLiteralContext: mut.toLiteralContext()
+            };
+    
+            
+
+            const to = window.setTimeout(() => {
+                if (this.pendingSyncStates.has(requestId)) {
+                    this.pendingSyncStates.delete(requestId)
+                    reject('timeout');
+                }
+            }, timeout);
+
+            this.pendingSyncStates.set(requestId, {resolve: resolve, reject: reject, timeout: to, mut: mut});
+
+            this.commandForwardingFn(cmd);
+        });
+
+        return p;
     }
 
     addSyncObserver(_obs: SyncObserver, _mut: MutableObject, _peerGroupId?: string | undefined): void {