Merge pull request #9 from ptcrealitylab/parallelSockets

Add support for creation of parallel sockets for large messages

Author: dangond-ptc

package.json

@@ -1,6 +1,6 @@
 {
   "name": "toolsocket",
-  "version": "2.0.1",
+  "version": "2.1.0",
   "description": "A communication layer built on top of WebSocket for Node.js and browsers.",
   "main": "src/index.js",
   "scripts": {

src/IncomingToolSocket.js

@@ -0,0 +1,27 @@
+const ToolSocket = require("./ToolSocket");
+
+class IncomingToolSocket extends ToolSocket {
+    /**
+     * Creates an IncomingToolSocket from an incoming WebSocket connection. Used by ToolSocketServer.
+     * @param {WebSocket} websocket - The incoming WebSocket connection.
+     * @param {ToolSocketServer} server - The ToolSocketServer this is attached to.
+     */
+    constructor(websocket, server) {
+        super();
+        this.socket = websocket;
+        this.networkId = 'toolbox'; // Or 'io'?
+        this.origin = server.origin;
+        this.server = server;
+        this.configureSocket();
+    }
+
+    /**
+     * Requests the source to create another ToolSocket connection for parallel data transfer.
+     * @return {Promise<ToolSocket>} - The parallel socket we just created.
+     */
+    requestParallelSocket() {
+        return this.server.requestParallelSocket(this);
+    }
+}
+
+module.exports = IncomingToolSocket;

src/ToolSocket.js

@@ -152,10 +152,20 @@ class ToolSocket {
             }
         });
 
+        this.addEventListener('meta', (route, body, _response, _binaryData, _messageBundle) => {
+            if (route === 'requestParallel') {
+                this.triggerEvent('requestParallel', body); // body = id
+            } else if (route === 'confirmParallel') {
+                this.triggerEvent('confirmParallel', body); // body = id
+            } else {
+                console.warn(`Received unknown meta route: "${route}"`);
+            }
+        });
+
         // We're receiving an event, trigger it
         this.addEventListener('io', (route, body, _responseObject, binaryData) => {
             if (VALID_METHODS.includes(route)) {
-                console.warn(`Received IO message with route, ${route}, which cannot be distinguished from the request method with the same name. Please pick a different route.`);
+                console.warn(`Received IO message with route: "${route}", which cannot be distinguished from the request method with the same name. Please pick a different route.`);
             }
             this.triggerEvent(route, body, binaryData);
         });
@@ -557,6 +567,17 @@ class ToolSocket {
         this.sendMethod('unsub', route, body, callback, binaryData);
     }
 
+    /**
+     * Sends a META message, used for ToolSocket internal messages (i.e. requestParallel)
+     * @param {string} route - The route
+     * @param {any} body - The message body
+     * @param {function} [callback] - A callback function that is called if a response is required
+     * @param {object} [binaryData] - Binary data
+     */
+    meta(route, body, callback, binaryData) {
+        this.sendMethod('meta', route, body, callback, binaryData);
+    }
+
     /**
      * Adds aliases for backwards compatibility
      */
@@ -570,6 +591,15 @@ class ToolSocket {
         this.CLOSING = WebSocketWrapper.CLOSING;
         this.CLOSED = WebSocketWrapper.CLOSED;
     }
+
+    /**
+     * Clones a ToolSocket, creating a parallel connection to the same endpoint.
+     * @param {ToolSocket} toolsocket - The source ToolSocket.
+     * @returns {ToolSocket} - A new ToolSocket created to the same endpoint as the original.
+     */
+    static makeParallelSocket(toolsocket) {
+        return new ToolSocket(toolsocket.url, toolsocket.networkId, 'parallel');
+    }
 }
 
 module.exports = ToolSocket;

src/ToolSocketServer.js

@@ -1,7 +1,7 @@
-const ToolSocket = require('./ToolSocket.js');
-
 const { WebSocketWrapper } = require('./utilities.js');
 const { URL_SCHEMA, MESSAGE_BUNDLE_SCHEMA } = require('./schemas.js');
+const IncomingToolSocket = require('./IncomingToolSocket');
+const {generateUniqueId} = require("./utilities");
 
 /**
  * A server for ToolSocket
@@ -21,19 +21,25 @@ class ToolSocketServer {
 
         this.eventCallbacks = {}; // For internal events
 
+        this.pendingParallelRequests = new Map();
+
         this.server.on('listening', (...args) => {
             this.triggerEvent('listening', ...args);
         });
 
         this.server.on('connection', socket => {
-            const toolSocket = new ToolSocket();
-            toolSocket.socket = socket;
-            toolSocket.networkId = 'toolbox'; // Or 'io'?
-            toolSocket.origin = this.origin;
-            toolSocket.configureSocket();
+            const toolSocket = new IncomingToolSocket(socket, this);
             this.sockets.push(toolSocket);
             this.triggerEvent('connection', toolSocket);
 
+            toolSocket.on('confirmParallel', id => {
+                if (this.pendingParallelRequests.has(id)) {
+                    const resolve = this.pendingParallelRequests.get(id);
+                    resolve(toolSocket);
+                    this.pendingParallelRequests.delete(id);
+                }
+            });
+
             socket.on('close', () => {
                 this.sockets.splice(this.sockets.indexOf(toolSocket), 1);
             });
@@ -88,6 +94,22 @@ class ToolSocketServer {
         this.server.server = this.server;
     }
 
+    /**
+     * Requests the source to create another ToolSocket connection for parallel data transfer.
+     * @param {ToolSocket} toolSocket - The toolSocket requesting a parallel connection.
+     * @return {Promise<ToolSocket>} - The parallel socket we just created.
+     */
+    requestParallelSocket(toolSocket) {
+        const id = generateUniqueId(8);
+        toolSocket.meta('requestParallel', id, null, null);
+        let resolve;
+        const promise = new Promise((res) => {
+            resolve = res;
+        });
+        this.pendingParallelRequests.set(id, resolve);
+        return promise;
+    }
+
     close() {
         this.server.close();
     }

src/constants.js

@@ -1,10 +1,10 @@
 const MAX_MESSAGE_SIZE = 300 * 1024 * 1024;
 const VALID_FILETYPES = ['css', 'csv', 'dat', 'fbx', 'gif', 'glb', 'htm', 'html', 'jpg', 'jpeg', 'js', 'json', 'map', 'mp4', 'obj', 'otf', 'pdf', 'ply', 'png', 'splat', 'svg', 'ttf', 'wasm', 'webm', 'webp', 'woff', 'xml', 'zip', '3dt'];
 /**
- * @typedef {'action' | 'beat' | 'delete' | 'get' | 'io' | 'keys' | 'message' | 'new' | 'patch' | 'ping' | 'post' | 'pub' | 'put' | 'res' | 'sub' | 'unsub'} MethodString
+ * @typedef {'action' | 'beat' | 'delete' | 'get' | 'io' | 'keys' | 'message' | 'new' | 'patch' | 'ping' | 'post' | 'pub' | 'put' | 'res' | 'sub' | 'unsub' | 'meta'} MethodString
  */
 /** @type MethodString[] */
-const VALID_METHODS = ['action', 'beat', 'delete', 'get', 'io', 'keys', 'message', 'new', 'patch', 'ping', 'post', 'pub', 'put', 'res', 'sub', 'unsub'];
+const VALID_METHODS = ['action', 'beat', 'delete', 'get', 'io', 'keys', 'message', 'new', 'patch', 'ping', 'post', 'pub', 'put', 'res', 'sub', 'unsub', 'meta'];
 
 module.exports = {
     MAX_MESSAGE_SIZE,

src/schemas.js

@@ -59,7 +59,7 @@ const MESSAGE_BUNDLE_SCHEMA = new Schema([
     ]),
     // Origin
     new Schema.StringValidator('o', {
-        enum: ['server', 'client', 'web', 'edge', 'proxy'],
+        enum: ['server', 'client', 'web', 'edge', 'proxy', 'parallel'],
         required: true
     }),
     // Network