added a "instanceId" field to connectionInfo in the NetworkAgent

this field is generated in the Linkup layer, and distinguishes from different devices that are listening on the same address (e.g. several devices owner by the same person, and using the same identity)

Author: sbazerque

src/mesh/agents/network/NetworkAgent.ts

@@ -8,7 +8,7 @@ import { WebRTCConnection } from 'net/transport/WebRTCConnection';
 import { RNGImpl } from 'crypto/random';
 import { SignallingServerConnection } from 'net/linkup/SignallingServerConnection';
 import { WebSocketConnection } from 'net/transport/WebSocketConnection';
-import { LinkupManagerHost, LinkupManagerEvent } from 'net/linkup';
+import { LinkupManagerHost, LinkupManagerEvent, NewCallMessageCallback } from 'net/linkup';
 import { WebRTCConnectionCommand, WebRTCConnectionEvent, WebRTCConnectionProxy } from 'net/transport';
 import { Identity } from 'data/identity';
 
@@ -78,6 +78,7 @@ type ConnectionInfo = {
     localEndpoint: Endpoint,
     remoteEndpoint: Endpoint, 
     connId : ConnectionId, 
+    remoteInstanceId?: string, // see the note in instanceIds in class SignallingServerConnection
     status: ConnectionStatus,
     timestamp: number,
     requestedBy: Set<AgentId>
@@ -121,13 +122,13 @@ class NetworkAgent implements Agent {
     connections  : Map<ConnectionId, Connection>;
 
     connectionInfo : Map<ConnectionId, ConnectionInfo>;
-    deferredInitialMessages : Map<ConnectionId, Array<any>>;
+    deferredInitialMessages : Map<ConnectionId, Array<{instanceId: string, message: any}>>;
 
     messageCallback : (data: any, conn: Connection) => void;
 
     connectionReadyCallback : (conn: Connection) => void;
 
-    newConnectionRequestCallback : (sender: LinkupAddress, receiver: LinkupAddress, callId: string, message: any) => void;
+    newConnectionRequestCallback : NewCallMessageCallback;
 
     linkupMessageCallback : (sender: LinkupAddress, receiver: LinkupAddress, message: any) => void;
 
@@ -209,6 +210,9 @@ class NetworkAgent implements Agent {
                 if (connInfo.status !== ConnectionStatus.Ready) {
                     this.connections.set(connectionId, conn);
                     connInfo.status = ConnectionStatus.Ready;
+                    if (connInfo.remoteInstanceId === undefined) {
+                        connInfo.remoteInstanceId = conn.remoteInstanceId;
+                    }
                     const ev: ConnectionStatusChangeEvent = {
                         type: NetworkEventType.ConnectionStatusChange, 
                         content: {
@@ -227,7 +231,7 @@ class NetworkAgent implements Agent {
             }
         }
 
-        this.newConnectionRequestCallback = (sender: LinkupAddress, receiver: LinkupAddress, connectionId: string, message: any) => {
+        this.newConnectionRequestCallback = (sender: LinkupAddress, receiver: LinkupAddress, connectionId: string, instanceId: string, message: any) => {
 
             let connInfo = this.connectionInfo.get(connectionId);
 
@@ -237,7 +241,8 @@ class NetworkAgent implements Agent {
                 connInfo = {
                     localEndpoint: receiver.url(),
                     remoteEndpoint: sender.url(), 
-                    connId: connectionId, 
+                    connId: connectionId,
+                    remoteInstanceId: instanceId,
                     status: ConnectionStatus.Received,
                     timestamp: Date.now(),
                     requestedBy: new Set()
@@ -246,13 +251,22 @@ class NetworkAgent implements Agent {
                 this.connectionInfo.set(connectionId, connInfo);
             }
 
+            /*if (connInfo.localEndpoint === receiver.url() &&
+                connInfo.remoteEndpoint === sender.url() &&
+                connInfo.remoteEndpoint !== instanceId) {
+
+                    console.log('MISMATCH')
+                    CONSOL
+            }*/
+
             if (connInfo.localEndpoint === receiver.url() &&
-                connInfo.remoteEndpoint === sender.url()) {
+                connInfo.remoteEndpoint === sender.url() &&
+                connInfo.remoteInstanceId === instanceId) {
 
                     if (connInfo.status === ConnectionStatus.Establishing) {
-                        this.acceptReceivedConnectionMessages(connectionId, message);
+                        this.acceptReceivedConnectionMessages(connectionId, instanceId, message);
                     } else if (connInfo.status === ConnectionStatus.Received) {
-                        this.deferReceivedConnectionMessage(connectionId, message);
+                        this.deferReceivedConnectionMessage(connectionId, instanceId, message);
 
                         if (isNew) {
                             let ev: ConnectionStatusChangeEvent = {
@@ -389,26 +403,27 @@ class NetworkAgent implements Agent {
     }
     */
 
-    private acceptReceivedConnectionMessages(connId: ConnectionId, message?: any) {
+    private acceptReceivedConnectionMessages(connId: ConnectionId, instanceId?: string, message?: any) {
 
         let messages = this.deferredInitialMessages.get(connId);
 
         if (messages === undefined) {
             messages = [];
         }
 
-        if (message !== undefined) {
-            messages.push(message);
+        if (message !== undefined && instanceId !== undefined) {
+            messages.push({instanceId: instanceId, message: message});
         }
 
 
-        for (const message of messages) {
+        for (const {message, instanceId } of messages) {
             let conn = this.connections.get(connId);
 
             if (conn === undefined) {
                 let connInfo = this.connectionInfo.get(connId) as ConnectionInfo;
 
                 if (connInfo !== undefined) {
+
                     const receiver = LinkupAddress.fromURL(connInfo.localEndpoint);
                     const sender   = LinkupAddress.fromURL(connInfo.remoteEndpoint);
 
@@ -434,13 +449,13 @@ class NetworkAgent implements Agent {
 
                 if (conn instanceof WebRTCConnection || conn instanceof WebRTCConnectionProxy || conn instanceof WebSocketConnection) {
                     conn.setMessageCallback(this.messageCallback);
-                    conn.answer(message);
+                    conn.answer(instanceId, message);
                 }
             } else {
                 if (conn instanceof WebRTCConnection || conn instanceof WebRTCConnectionProxy) {
-                    conn.receiveSignallingMessage(message);
+                    conn.receiveSignallingMessage(instanceId, message);
                 } else if (conn instanceof WebSocketConnection) {
-                    conn.answer(message);
+                    conn.answer(instanceId, message);
                 }
             }
 
@@ -450,16 +465,16 @@ class NetworkAgent implements Agent {
         }
     }
 
-    private deferReceivedConnectionMessage(connId: ConnectionId, message: any) {
+    private deferReceivedConnectionMessage(connId: ConnectionId, instanceId: string, message: any) {
 
         let messages = this.deferredInitialMessages.get(connId);
 
         if (messages === undefined) {
-            messages = new Array<any>();
+            messages = new Array<{instanceId: string, message: any}>();
             this.deferredInitialMessages.set(connId, messages);
         }
 
-        messages.push(message);
+        messages.push({message: message, instanceId: instanceId});
     }
 
     // Network listen, shutdown
@@ -534,7 +549,8 @@ class NetworkAgent implements Agent {
             { 
                 localEndpoint: local, 
                 remoteEndpoint: remote, 
-                connId: callId, status: ConnectionStatus.Establishing, 
+                connId: callId, 
+                status: ConnectionStatus.Establishing, 
                 timestamp: Date.now(),
                 requestedBy: new Set([requestedBy])
             });

src/mesh/agents/peer/PeerGroupAgent.ts

@@ -452,6 +452,12 @@ class PeerGroupAgent implements Agent {
 
     // Connection deduplication logic.
 
+    // Note: ATM the PeerGroupAgent will aggressively deduplicate all connections that go to the
+    //       same endpoint. In the future, we could use the "instanceId" field in the NetworkAgent's
+    //       connectionInfo to tell when these connections are actually going to the same devices 
+    //       (besides belonging to the same identity), and may not prune connections up to a given
+    //       number of different devices.
+
     private deduplicateConnections() {
 
         for (const [endpoint, connIds] of this.connectionsPerEndpoint.entries()) {

src/net/linkup/LinkupServer.ts

@@ -1,7 +1,7 @@
 import { LinkupAddress } from './LinkupAddress';
 
-type NewCallMessageCallback = (sender: LinkupAddress, recipient: LinkupAddress, callId: string, message: any) => void;
-type MessageCallback        = (message: any) => void;
+type NewCallMessageCallback = (sender: LinkupAddress, recipient: LinkupAddress, callId: string, instanceId: string, message: any) => void;
+type MessageCallback        = (instanceId: string, message: any) => void;
 
 type ListeningAddressesQueryCallback  = (queryId: string, matches: Array<LinkupAddress>) => void;
 

src/net/linkup/SignallingServerConnection.ts

@@ -2,9 +2,20 @@ import { Logger, LogLevel } from 'util/logging';
 
 import { LinkupServer, RawMessageCallback, NewCallMessageCallback, MessageCallback, ListeningAddressesQueryCallback } from './LinkupServer';
 import { LinkupAddress } from './LinkupAddress';
+import { RNGImpl } from 'crypto/random';
 
 const CONN_BACKOFF_TIME = 15000;
 
+/*
+ * On instanceIds: when connecting to a linkup server, the SignallingServerConnection class will
+ *                 create a randm "instanceId" value. When several peers are listening for 
+ *                 connections on the same endpoint (e.g. several devices owner by the same person) 
+ *                 and another peer initiates a connection, each of these peers will include a
+ *                 different instanceId. This enables the initiator to tell them appart in the
+ *                 WebRTC signalling phase, and to choose if he wants to connect to just one of
+ *                 them, a few, etc.
+ */
+
 class SignallingServerConnection implements LinkupServer {
 
     static logger = new Logger(SignallingServerConnection.name, LogLevel.ERROR);
@@ -19,7 +30,8 @@ class SignallingServerConnection implements LinkupServer {
         return serverURL.slice(SignallingServerConnection.WRTC_URL_PREFIX.length);
     }
 
-    readonly serverURL : string;
+    readonly serverURL  : string;
+    readonly instanceId : string; // see note above
 
     ws : WebSocket | null;
 
@@ -44,6 +56,7 @@ class SignallingServerConnection implements LinkupServer {
         }
 
         this.serverURL = serverURL;
+        this.instanceId  = new RNGImpl().randomHexString(128);
 
         this.ws = null;
 
@@ -153,6 +166,7 @@ class SignallingServerConnection implements LinkupServer {
                     'action'         :  'send',
                     'linkupId'       :  recipient.linkupId,
                     'callId'         :  callId,
+                    'instanceId'     :  this.instanceId,
                     'data'           :  data,
                     'replyServerUrl' :  sender.serverURL,
                     'replyLinkupId'  :  sender.linkupId,
@@ -269,7 +283,7 @@ class SignallingServerConnection implements LinkupServer {
                                     if (callMessageCallbacks !== undefined) {
                                         callMessageCallbacks.forEach((callback: MessageCallback) => {
                                             SignallingServerConnection.logger.debug('Delivering linkup message to ' + linkupId + ' on call ' + message['callId']);
-                                            callback(message['data']);
+                                            callback(message['instanceId'], message['data']);
                                             found = true;
                                         });
                                     }
@@ -281,7 +295,7 @@ class SignallingServerConnection implements LinkupServer {
                                     if (linkupIdCallbacks !== undefined) {
                                         linkupIdCallbacks.forEach((callback: NewCallMessageCallback) => {
                                             SignallingServerConnection.logger.debug('Calling default callback for linkupId ' + linkupId + ', unlistened callId is ' + callId);
-                                            callback(new LinkupAddress(message['replyServerUrl'], message['replyLinkupId']), new LinkupAddress(this.serverURL, linkupId), callId, message['data']);
+                                            callback(new LinkupAddress(message['replyServerUrl'], message['replyLinkupId']), new LinkupAddress(this.serverURL, linkupId), callId, message['instanceId'], message['data']);
                                             found = true;
                                         })
                                     }

src/net/linkup/WebSocketListener.ts

@@ -89,7 +89,7 @@ class WebSocketListener implements LinkupServer {
 
                         if (callbacks.size > 0) {
                             for (const callback of callbacks) {
-                                callback(sender, recipient, connId, {ws: socket, reverse: reverse});
+                                callback(sender, recipient, connId, undefined as any, {ws: socket, reverse: reverse});
                                 parseOK = true;
                             }
                         } else {

src/net/linkup/remoting/LinkupManagerHost.ts

@@ -64,6 +64,7 @@ type NewCallMessageEvent = {
     sender: Endpoint,
     recipient: Endpoint,
     callId: string,
+    instanceId: string
     message: any
 }
 
@@ -84,6 +85,7 @@ type MessageOnCallEvent = {
     type: 'message-on-call',
     recipient: Endpoint,
     callId: string,
+    instanceId: string,
     message: any
 }
 
@@ -128,13 +130,14 @@ class LinkupManagerHost {
         this.eventCallback = eventCallback;
 
         this.newCallMessageCallback = 
-            (sender: LinkupAddress, recipient: LinkupAddress, callId: string, message: any) => {
+            (sender: LinkupAddress, recipient: LinkupAddress, callId: string, instanceId: string, message: any) => {
 
                 const ev: NewCallMessageEvent = {
                     type: 'new-call-message',
                     sender: sender.url(),
                     recipient: recipient.url(),
                     callId: callId,
+                    instanceId: instanceId,
                     message: message
                 }
 
@@ -186,11 +189,12 @@ class LinkupManagerHost {
 
             const listen = cmd as ListenForMessagesOnCall;
 
-            const callback = (msg: any) => {
+            const callback: MessageCallback = (instanceId: string, msg: any) => {
                 const ev: MessageOnCallEvent = {
                     type: 'message-on-call',
                     recipient: listen.recipient,
                     callId: listen.callId,
+                    instanceId: instanceId,
                     message: msg
                 };
 

src/net/linkup/remoting/LinkupManagerProxy.ts

@@ -40,7 +40,7 @@ class LinkupManagerProxy {
                     const recipient = LinkupAddress.fromURL(newCall.recipient);
 
                     try {
-                        cb(sender, recipient, newCall.callId, newCall.message);
+                        cb(sender, recipient, newCall.callId, newCall.instanceId, newCall.message);
                     } catch (e) {
                         console.log('Error in callback invocation within LinkupManagerProxy: ', e);
                     }
@@ -82,7 +82,7 @@ class LinkupManagerProxy {
                 for (const cb of this.messagesOnCallCallbacks.get(msg.recipient + '/' + msg.callId)) {
 
                     try {
-                        cb(msg.message);
+                        cb(msg.instanceId, msg.message);
                     } catch (e) {
                         console.log('Error in callback invocation within LinkupManagerProxy: ', e);
                     }

src/net/transport/Connection.ts

@@ -4,6 +4,7 @@ interface Connection {
 
     readonly localAddress: LinkupAddress;
     readonly remoteAddress: LinkupAddress;
+    readonly remoteInstanceId?: string;
 
     getConnectionId() : string;
     initiatedLocally(): boolean;