Update the pads to run with the latest improvements to the websocket server

NetfluxWebsocketSrv.js

@@ -0,0 +1,198 @@
+;(function () { 'use strict';
+const Crypto = require('crypto');
+const LogStore = require('./storage/LogStore');
+
+const LAG_MAX_BEFORE_DISCONNECT = 30000;
+const LAG_MAX_BEFORE_PING = 15000;
+const HISTORY_KEEPER_ID = Crypto.randomBytes(8).toString('hex');
+
+const USE_HISTORY_KEEPER = true;
+
+let dropUser;
+
+const now = function () { return (new Date()).getTime(); };
+
+const sendMsg = function (ctx, user, msg) {
+    try {
+        console.log('<' + JSON.stringify(msg));
+        user.socket.send(JSON.stringify(msg));
+    } catch (e) {
+        console.log(e.stack);
+        dropUser(ctx, user);
+    }
+};
+
+const sendChannelMessage = function (ctx, channel, msgStruct) {
+    msgStruct.unshift(0);
+    channel.forEach(function (user) {
+      if(msgStruct[2] !== 'MSG' || user.id !== msgStruct[1]) { // We don't want to send back a message to its sender, in order to save bandwidth
+        sendMsg(ctx, user, msgStruct);
+      }
+    });
+    if (USE_HISTORY_KEEPER && msgStruct[2] === 'MSG') {
+        ctx.store.message(channel.id, JSON.stringify(msgStruct), function () { });
+    }
+};
+
+dropUser = function (ctx, user) {
+    if (user.socket.readyState !== 2 /* WebSocket.CLOSING */
+        && user.socket.readyState !== 3 /* WebSocket.CLOSED */)
+    {
+        try {
+            user.socket.close();
+        } catch (e) {
+            console.log("Failed to disconnect ["+user.id+"], attempting to terminate");
+            try {
+                user.socket.terminate();
+            } catch (ee) {
+                console.log("Failed to terminate ["+user.id+"]  *shrug*");
+            }
+        }
+    }
+    delete ctx.users[user.id];
+    Object.keys(ctx.channels).forEach(function (chanName) {
+        let chan = ctx.channels[chanName];
+        let idx = chan.indexOf(user);
+        if (idx < 0) { return; }
+        console.log("Removing ["+user.id+"] from channel ["+chanName+"]");
+        chan.splice(idx, 1);
+        if (chan.length === 0) {
+            console.log("Removing empty channel ["+chanName+"]");
+            delete ctx.channels[chanName];
+        } else {
+            sendChannelMessage(ctx, chan, [user.id, 'LEAVE', chanName, 'Quit: [ dropUser() ]']);
+        }
+    });
+};
+
+const getHistory = function (ctx, channelName, handler) {
+    ctx.store.getMessages(channelName, function (msgStr) { handler(JSON.parse(msgStr)); });
+};
+
+const randName = function () { return Crypto.randomBytes(16).toString('hex'); };
+
+const handleMessage = function (ctx, user, msg) {
+    let json = JSON.parse(msg);
+    let seq = json.shift();
+    let cmd = json[0];
+    let obj = json[1];
+
+    user.timeOfLastMessage = now();
+    user.pingOutstanding = false;
+
+    if (cmd === 'JOIN') {
+        if (obj && obj.length !== 32) {
+            sendMsg(ctx, user, [seq, 'ERROR', 'ENOENT', obj]);
+            return;
+        }
+        let chanName = obj || randName();
+        sendMsg(ctx, user, [seq, 'ACK', chanName]);
+        let chan = ctx.channels[chanName] = ctx.channels[chanName] || [];
+        chan.id = chanName;
+        if (USE_HISTORY_KEEPER) {
+            sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'JOIN', chanName]);
+        }
+        chan.forEach(function (u) { sendMsg(ctx, user, [0, u.id, 'JOIN', chanName]); });
+        chan.push(user);
+        sendChannelMessage(ctx, chan, [user.id, 'JOIN', chanName]);
+        return;
+    }
+    if (cmd === 'MSG') {
+        if (obj === HISTORY_KEEPER_ID) {
+            let parsed;
+            try { parsed = JSON.parse(json[2]); } catch (err) { return; }
+            if (parsed[0] === 'GET_HISTORY') {
+                console.log('getHistory ' + parsed[1]);
+                getHistory(ctx, parsed[1], function (msg) {
+                    sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, JSON.stringify(msg)]);
+                });
+                sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'MSG', user.id, 0]);
+            }
+            return;
+        }
+        if (obj && !ctx.channels[obj] && !ctx.users[obj]) {
+            sendMsg(ctx, user, [seq, 'ERROR', 'ENOENT', obj]);
+            return;
+        }
+        sendMsg(ctx, user, [seq, 'ACK', '']);
+        let target;
+        json.unshift(user.id);
+        if ((target = ctx.channels[obj])) {
+            sendChannelMessage(ctx, target, json);
+            return;
+        }
+        if ((target = ctx.users[obj])) {
+            json.unshift(0);
+            sendMsg(ctx, target, json);
+            return;
+        }
+    }
+    if (cmd === 'LEAVE') {
+        let err;
+        let chan;
+        let idx;
+        if (!obj) { err = 'EINVAL'; }
+        if (!err && !(chan = ctx.channels[obj])) { err = 'ENOENT'; }
+        if (!err && (idx = chan.indexOf(user)) === -1) { err = 'NOT_IN_CHAN'; }
+        if (err) {
+            sendMsg(ctx, user, [seq, 'ERROR', err]);
+            return;
+        }
+        sendMsg(ctx, user, [seq, 'ACK', chan.id]);
+        json.unshift(user.id);
+        sendChannelMessage(ctx, chan, [user.id, 'LEAVE', chan.id]);
+        chan.splice(idx, 1);
+    }
+    if (cmd === 'PING') {
+        sendMsg(ctx, user, [seq, 'ACK', obj]);
+        return;
+    }
+};
+
+let run = module.exports.run = function (storage, socketServer) {
+    let ctx = {
+        users: {},
+        channels: {},
+        store: LogStore.create('messages.log', storage)
+    };
+    setInterval(function () {
+        Object.keys(ctx.users).forEach(function (userId) {
+            let u = ctx.users[userId];
+            if (now() - u.timeOfLastMessage > LAG_MAX_BEFORE_DISCONNECT) {
+                dropUser(ctx, u);
+            } else if (!u.pingOutstanding && now() - u.timeOfLastMessage > LAG_MAX_BEFORE_PING) {
+                sendMsg(ctx, u, [0, 'PING', now()]);
+                u.pingOutstanding = true;
+            }
+        });
+    }, 5000);
+    socketServer.on('connection', function(socket) {
+        let conn = socket.upgradeReq.connection;
+        let user = {
+            addr: conn.remoteAddress + '|' + conn.remotePort,
+            socket: socket,
+            id: randName(),
+            timeOfLastMessage: now(),
+            pingOutstanding: false
+        };
+        ctx.users[user.id] = user;
+        sendMsg(ctx, user, [0, '', 'IDENT', user.id]);
+        socket.on('message', function(message) {
+            console.log('>'+message);
+            try {
+                handleMessage(ctx, user, message);
+            } catch (e) {
+                console.log(e.stack);
+                dropUser(ctx, user);
+            }
+        });
+        socket.on('close', function (evt) {
+            for (let userId in ctx.users) {
+                if (ctx.users[userId].socket === socket) {
+                    dropUser(ctx, ctx.users[userId]);
+                }
+            }
+        });
+    });
+};
+}());

customize.dist/index.html

@@ -125,9 +125,9 @@
         <script>
             require(['/common/crypto.js', '/api/config?cb=' + Math.random().toString(16).substring(2)], function (Crypto, Config) {
                 document.getElementById('buttons').setAttribute('style', '');
-                document.getElementById('create-pad').setAttribute('href', '/pad/#' + Crypto.genKey());
+                document.getElementById('create-pad').setAttribute('href', '/pad/');
                 if(Config.webrtcURL !== '') {
-                  document.getElementById('create-rtcpad').setAttribute('href', '/pad/?webrtc=1#' + Crypto.genKey());
+                  document.getElementById('create-rtcpad').setAttribute('href', '/pad/?webrtc=1');
                 }
                 document.getElementById('create-sheet').setAttribute('href', '/sheet/#' + Crypto.genKey());
                 document.getElementById('create-code').setAttribute('href', '/code/#' + Crypto.genKey());

server.js

@@ -7,7 +7,8 @@ var Https = require('https');
 var Fs = require('fs');
 var WebSocketServer = require('ws').Server;
 var ChainPadSrv = require('./ChainPadSrv');
-var NetfluxSrv = require('./NetFluxWebsocketServer');
+// var NetfluxSrv = require('./NetFluxWebsocketServer');
+var NetfluxSrv = require('./NetfluxWebsocketSrv');
 var WebRTCSrv = require('./WebRTCSrv');
 
 var config = require('./config');

www/common/netflux.js

@@ -675,10 +675,16 @@ return /******/ (function(modules) { // webpackBootstrap
 	            var msg = undefined;
 	            if (data.type === 'PING') {
 	              var date = new Date().getTime();
-	              // webChannel.lastPing = date;
 	              msg = JSON.stringify([0, 'PING', date]);
 	            } else {
 	              msg = JSON.stringify([c.seq++, data.type, webChannel.id, data.msg]);
+	              if (data.type === 'MSG') {
+	                var srvMsg = JSON.parse(msg);
+	                srvMsg.shift();
+	                srvMsg.unshift(webChannel.myID);
+	                srvMsg.unshift(0);
+	                webChannel.waitingAck[c.seq - 1] = srvMsg;
+	              }
 	            }
 	            c.send(msg);
 	          }
@@ -1316,15 +1322,16 @@ return /******/ (function(modules) { // webpackBootstrap
 	      var webChannel = socket.webChannel;
 	      var topology = cs.STAR_SERVICE;
 	      var topologyService = _ServiceProvider2.default.get(topology);
-	      var HISTORY_KEEPER = '_HISTORY_KEEPER_';
+	      var history_keeper = webChannel.hc;
 
-	      if (msg[0] !== 0) {
+	      if (msg[0] !== 0 && msg[1] !== 'ACK') {
 	        return;
 	      }
-	      if (msg[1] === 'IDENT') {
+	      if (msg[2] === 'IDENT' && msg[1] === '') {
-	        socket.uid = msg[2];
+	        socket.uid = msg[3];
-	        webChannel.myID = msg[2];
+	        webChannel.myID = msg[3];
 	        webChannel.peers = [];
+	        webChannel.waitingAck = [];
 	        webChannel.topology = topology;
 	        return;
 	      }
@@ -1333,17 +1340,25 @@ return /******/ (function(modules) { // webpackBootstrap
 	        socket.send(JSON.stringify(msg));
 	        return;
 	      }
-	      if (msg[1] === 'PONG') {
+	      if (msg[1] === 'ACK' && parseInt(msg[2]) === msg[2]) {
 	        var lag = new Date().getTime() - msg[2];
 	        webChannel.getLag = function () {
 	          return lag;
 	        };
 	        return;
 	      }
+	      if (msg[1] === 'ACK') {
+	        var seq = msg[0];
+	        if (webChannel.waitingAck[seq]) {
+	          var newMsg = webChannel.waitingAck[seq];
+	          delete webChannel.waitingAck[seq];
+	          if (typeof webChannel.onmessage === "function") webChannel.onmessage(newMsg[1], newMsg[4]);
+	        }
+	      }
 	      // We have received a new direct message from another user
 	      if (msg[2] === 'MSG' && msg[3] === socket.uid) {
-	        // If it comes form the history keeper, send it to the user
+	        // If it comes from the history keeper, send it to the user
-	        if (msg[1] === HISTORY_KEEPER) {
+	        if (msg[1] === history_keeper) {
 	          if (msg[4] === 0) {
 	            webChannel.onmessage(msg[1], msg[4]);
 	            return;
@@ -1367,7 +1382,12 @@ return /******/ (function(modules) { // webpackBootstrap
 	        } else {
 	          // Trigger onJoining() when another user is joining the channel
 	          // Register the user in the list of peers in the channel
-	          var linkQuality = msg[1] === HISTORY_KEEPER ? 1000 : 0;
+	          if (webChannel.peers.length === 0 && msg[1].length === 16) {
+	            // We've just catched the history keeper
+	            history_keeper = msg[1];
+	            webChannel.hc = history_keeper;
+	          }
+	          var linkQuality = msg[1] === history_keeper ? 1000 : 0;
 	          var sendToPeer = function sendToPeer(data) {
 	            topologyService.sendTo(msg[1], webChannel, { type: 'MSG', msg: data });
 	          };
@@ -1376,7 +1396,7 @@ return /******/ (function(modules) { // webpackBootstrap
 	            webChannel.peers.push(peer);
 	          }
 
-	          if (msg[1] !== HISTORY_KEEPER) {
+	          if (msg[1] !== history_keeper) {
 	            // Trigger onJoining with that peer once the function is loaded (i.e. once the channel is synced)
 	            var waitForOnJoining = function waitForOnJoining() {
 	              if (typeof webChannel.onJoining !== "function") {

www/common/realtime-input.js

@@ -78,7 +78,8 @@ define([
         var webrtcUrl = config.webrtcURL;
         var userName = config.userName;
         var channel = config.channel;
-        var cryptKey = config.cryptKey;
+        var chanKey = config.cryptKey;
+        var cryptKey = Crypto.parseKey(chanKey).cryptKey;
         var passwd = 'y';
 
         // make sure configuration is defined
@@ -114,10 +115,10 @@ define([
             return res;
         };
 
-        var mkMessage = function (user, channel, content) {
+        var mkMessage = function (user, chan, content) {
             content = JSON.stringify(content);
             return user.length + ':' + user +
-                channel.length + ':' + channel +
+                chan.length + ':' + chan +
                 content.length + ':' + content;
         };
 
@@ -223,7 +224,7 @@ define([
         };
 
         var options = {
-          key: channel
+          key: ''
         };
 
         var rtc = true;
@@ -240,7 +241,7 @@ define([
           options.signaling = webrtcUrl;
         }
 
-        var createRealtime = function() {
+        var createRealtime = function(chan) {
             return ChainPad.create(userName,
                                         passwd,
                                         channel,
@@ -274,6 +275,8 @@ define([
         }
 
         var onOpen = function(wc) {
+            channel = wc.id;
+            window.location.hash = channel + '|' + chanKey;
             // Add the handlers to the WebChannel
             wc.onmessage = function(peer, msg) { // On receiving message
                 onMessage(peer, msg, wc);

www/pad/main.js

@@ -44,16 +44,24 @@ define([
     };
 
     var andThen = function (Ckeditor) {
-        $(window).on('hashchange', function() {
+        // $(window).on('hashchange', function() {
-            window.location.reload();
+            // window.location.reload();
-        });
+        // });
+        var key;
+        var channel = '';
         if (window.location.href.indexOf('#') === -1) {
-            window.location.href = window.location.href + '#' + Crypto.genKey();
+            key = Crypto.genKey();
-            return;
+            // window.location.href = window.location.href + '#' + Crypto.genKey();
+            // return;
+        }
+        else {
+            var hash = window.location.hash.substring(1);
+            channel = hash.substr(0,32);
+            key = hash.substr(33);
         }
 
         var fixThings = false;
-        var key = Crypto.parseKey(window.location.hash.substring(1));
+        // var key = Crypto.parseKey(window.location.hash.substring(1));
         var editor = window.editor = Ckeditor.replace('editor1', {
             // https://dev.ckeditor.com/ticket/10907
             needsBrFiller: fixThings,
@@ -220,10 +228,10 @@ define([
                 userName: userName,
 
                 // the channel we will communicate over
-                channel: key.channel,
+                channel: channel,
 
                 // our encryption key
-                cryptKey: key.cryptKey,
+                cryptKey: key,
 
                 // configuration :D
                 doc: inner,