Full WebRTC Support (#40)

louisgv · ColdSauce · commit e46883e8e842 · 2018-04-24T00:02:02.000-04:00
* WIP

* Adding data event handler on browser

* WIP

* WIP

* Implement simple if-else on join

* derp

* Need to investigate where did browser got data

* Got the data

* Adding clean up events

* Remove unnecessary code.

* Remove hard coded process env local.

* Remove TODOS and commented out code.
diff --git a/packages/hyperproxy-browser/public/index.html b/packages/hyperproxy-browser/public/index.html
@@ -68,8 +68,8 @@ <h2>
 
     <section id="browser"
              class="enabled">
-        <div class="hidden">
-            <h2>UID-<span id="swarm-id" class="light-font"></span></h2>
+        <div>
+            <h2 class="hidden">UID-<span id="swarm-id" class="light-font"></span></h2>
             <!-- <h3><span class="bold-font">URL</span><span id="active-daturl" class="light-font"></span></h3> -->
         </div>
 
diff --git a/packages/hyperproxy-browser/src/index.js b/packages/hyperproxy-browser/src/index.js
@@ -19,6 +19,30 @@ function init() {
     const urlFormEl = document.querySelector('#url-form');
     const contentEl = document.querySelector('#content');
 
+    const originalOnMessage = client.hub.onMessage;
+
+    window.addEventListener('unload', ()=>{
+        client.hub.broadcast(activeHash, {
+            from: client.swarm.me,
+            type: HUB_MSG_TYPE.LEAVE
+        });
+    });
+
+    client.swarm.on('peer', function(peer) {
+        peer.on('data', function(data) {
+            renderOutput(data);
+        });
+    });
+
+    function renderOutput(data) {
+        const {body, type, from} = JSON.parse(Buffer.from(data).toString('utf8'));
+        const value = body.type === 'Buffer'
+            ? Buffer.from(body.data).toString('utf8')
+            : body.data;
+
+        contentEl.innerHTML = value;
+    }
+
     document.querySelector('#swarm-id').innerHTML = client.swarm.me;
 
     document.querySelector('#subscribe').addEventListener('click', () => {
@@ -62,7 +86,7 @@ function init() {
         });
     });
 
-    document.querySelector('#try-out').addEventListener('click', ()=>{
+    document.querySelector('#try-out').addEventListener('click', () => {
         introEl.classList.remove('enabled');
     });
 }
diff --git a/packages/hyperproxy-config/index.js b/packages/hyperproxy-config/index.js
@@ -6,7 +6,6 @@ const HUB_NAME = 'hyperproxy';
 
 const HUB_IP = process.env.HUB_IP || 'localhost';
 const HUB_PORT = process.env.HUB_PORT || 9999;
-
 const HUB_URL = process.env.LOCAL
     ? `ws://${HUB_IP}:${HUB_PORT}/`
     : `wss://hub.hyperproxy.network/`;
@@ -18,8 +17,8 @@ module.exports = Object.freeze({
     HUB_URL,
     HUB_MSG_TYPE: {
         JOIN: 'JOIN',
+        LEAVE: 'LEAVE',
         REQUEST: 'REQUEST',
         RESPONSE: 'RESPONSE',
-        DISCONNECT: 'DISCONNECT'
     }
 });
diff --git a/packages/hyperproxy-hub-client/src/index.js b/packages/hyperproxy-hub-client/src/index.js
@@ -2,7 +2,7 @@ import swarm from 'webrtc-swarm';
 import signalhub from 'signalhubws';
 
 import HyperProxyLogger from 'hyperproxy-logger';
-import {HUB_NAME, HUB_URL} from 'hyperproxy-config';
+import { HUB_NAME, HUB_URL } from 'hyperproxy-config';
 
 const logger = new HyperProxyLogger('hyperproxy-hub');
 
@@ -11,10 +11,10 @@ export default class HyperproxyHubClient {
     constructor(wsClass, wrtc = null, name = HUB_NAME, url = HUB_URL) {
         this.hub = signalhub(name, [url], wsClass);
         this.swarm = wrtc
-            ? swarm(this.hub, {wrtc})
+            ? swarm(this.hub, { wrtc })
             : swarm(this.hub);
 
-        this.handleSwarmEvent();
+            this.handleSwarmEvent();
     }
 
     /* Waiting Coroutine */
@@ -33,20 +33,20 @@ export default class HyperproxyHubClient {
         const sw = this.swarm;
 
         /* Swarm Event */
-        sw.on('close', function(e) {
+        sw.on('close', function (e) {
             sw.close();
         });
 
-        sw.on('message', function(m) {
+        sw.on('message', function (m) {
             logger.info(m, 'new swarm message');
         });
 
-        sw.on('peer', function(peer, id) {
+        sw.on('peer', function (peer, id) {
             logger.info(id, 'peer connected');
             logger.info(sw.peers.length, 'total peers');
         });
 
-        sw.on('disconnect', function(peer, id) {
+        sw.on('disconnect', function (peer, id) {
             logger.info(id, 'peer disconnected');
             logger.info(sw.peers.length, 'total peers');
         });
diff --git a/packages/hyperproxy-hub/package.json b/packages/hyperproxy-hub/package.json
@@ -7,8 +7,9 @@
     "hyperproxy-hub": "./index.js"
   },
   "scripts": {
+    "dev": "devtool -r babel-register src/index.js",
     "start": "node lib/index.js",
-    "watch": "babel-watch src/index.js",
+    "watch": "babel-watch --inspect src/index.js",
     "build": "babel src -d lib",
     "test": "echo \"Error: no test specified\" && exit 1"
   },
diff --git a/packages/hyperproxy-hub/src/index.js b/packages/hyperproxy-hub/src/index.js
@@ -5,6 +5,7 @@ import HyperProxyNode from 'hyperproxy-node';
 import HyperProxyLogger from 'hyperproxy-logger';
 
 const datMap = new Map();
+const nodeMap = new Map();
 
 const wss = new WebSocket.Server({port});
 
@@ -28,24 +29,49 @@ function onConnection(ws) {
             return;
         }
 
-        logger.info(jsond, 'new websocket message');
-
         // if not all channel and no HyperProxy-node has been assigned
-        if (jsond.channel !== 'all' && jsond.message.type === HUB_MSG_TYPE.JOIN) {
-            if (!datMap[jsond.channel]) {
-                logger.info(jsond.channel, 'spawn new hyperproxy-node');
-                datMap[jsond.channel] = new HyperProxyNode(jsond.channel);
+        if (jsond.channel !== 'all') {
+            logger.info(jsond, 'Got message');
+
+            if (jsond.message.type === HUB_MSG_TYPE.JOIN) {
+                logger.info('received a JOIN event');
+
+                if (!datMap[jsond.channel]) {
+                    logger.info(jsond.channel, 'spawn new hyperproxy-node');
+                    const nodeInstance = new HyperProxyNode(jsond.channel);
+
+                    //TODO: Extends this to be a more generic object that has peer ID
+                    datMap[jsond.channel] = {
+                        [nodeInstance.client.swarm.me]: true,
+                        [jsond.message.from]: true
+                    };
+                    nodeMap[jsond.channel] = nodeInstance;
+                } else {
+                    datMap[jsond.channel][jsond.message.from] = true;
+                    return;
+                }
             }
 
-            return;
-        }
+            if (jsond.message.type === HUB_MSG_TYPE.LEAVE) {
+                logger.info('received a LEAVE event');
+                delete datMap[jsond.channel][jsond.message.from];
+                if (Object.keys(datMap[jsond.channel]).length === 1) {
+                    nodeMap[jsond.channel].close();
+                    delete nodeMap[jsond.channel];
+                    delete datMap[jsond.channel];
+                }
 
-        wss.clients.forEach((client) => {
-            if (jsond.app === client.app) {
-                logger.info(client.app, 'broadcasting');
-                client.send(data);
             }
-        });
+        }
+
+        wss
+            .clients
+            .forEach((client) => {
+                if (jsond.app === client.app) {
+                    logger.info(client.app, 'broadcasting');
+                    client.send(data);
+                }
+            });
     });
 }
 
diff --git a/packages/hyperproxy-node/src/index.js b/packages/hyperproxy-node/src/index.js
@@ -21,6 +21,11 @@ export default class HyperproxyNode {
         }
     }
 
+    close(){
+        this.client.swarm.close();
+        this.client.hub.close();
+    }
+
     /*
         Read a file from the subscribed dat archive asyncornously
     */
@@ -40,11 +45,11 @@ export default class HyperproxyNode {
     /*
         Connect to the hub under the specified key.
     */
-    async _connectToHub(key) {
-        const client = new HyperproxyHubClient(WebSocket, Wrtc);
-        client.hub.subscribe(key).on('data', (data) => {
+    _connectToHub(key) {
+        this.client = new HyperproxyHubClient(WebSocket, Wrtc);
+        this.client.hub.subscribe(key).on('data', (data) => {
             try {
-                this._handleData(client, key, data);
+                this._handleData(key, data);
             } catch (e) {
                 logger.error(e, 'cannot handle data');
             }
@@ -54,18 +59,21 @@ export default class HyperproxyNode {
     /*
         Handle any peer data by getting them from TCP/UDP dat
     */
-    async _handleData(client, key, {from, body, type}) {
+    async _handleData(key, {from, body, type}) {
         if (type === HUB_MSG_TYPE.RESPONSE) {
             return;
         }
 
         if (type === HUB_MSG_TYPE.REQUEST) {
+
             const file = await this.readFile(body);
-            client.hub.broadcast(key, {
-                from: client.swarm.me,
+            const payload = JSON.stringify({
+                from: this.client.swarm.me,
                 type: HUB_MSG_TYPE.RESPONSE,
                 body: file
             });
+
+            this.client.swarm.remotes[from].send(Buffer.from(payload));
         }
     }
 
@@ -92,4 +100,5 @@ export default class HyperproxyNode {
             });
         });
     }
-}
+}
+
