Merge pull request #21 from xwiki-labs/remove-channels

Remove empty channels

NetfluxWebsocketSrv.js

@@ -2,13 +2,13 @@
 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;
 const USE_FILE_BACKUP_STORAGE = true;
-const LOG_MESSAGES = false;
 
 
 let dropUser;
@@ -17,7 +17,7 @@ const now = function () { return (new Date()).getTime(); };
 
 const sendMsg = function (ctx, user, msg) {
     try {
-        if (LOG_MESSAGES) { console.log('<' + JSON.stringify(msg)); }
+        if (ctx.config.logToStdout) { console.log('<' + JSON.stringify(msg)); }
         user.socket.send(JSON.stringify(msg));
     } catch (e) {
         console.log(e.stack);
@@ -62,6 +62,24 @@ dropUser = function (ctx, user) {
         if (chan.length === 0) {
             console.log("Removing empty channel ["+chanName+"]");
             delete ctx.channels[chanName];
+
+            /*  Call removeChannel if it is a function and channel removal is
+                set to true in the config file */
+            if (ctx.config.removeChannels) {
+                if (typeof(ctx.store.removeChannel) === 'function') {
+                    ctx.timeouts[chanName] = setTimeout(function () {
+                        ctx.store.removeChannel(chanName, function (err) {
+                            if (err) { console.error("[removeChannelErr]: %s", err); }
+                            else {
+                                console.log("Deleted channel [%s] history from database...", chanName);
+                            }
+                        });
+                    }, ctx.config.channelRemovalTimeout);
+                } else {
+                    console.error("You have configured your server to remove empty channels, " +
+                        "however, the database adaptor you are using has not implemented this behaviour.");
+                }
+            }
         } else {
             sendChannelMessage(ctx, chan, [user.id, 'LEAVE', chanName, 'Quit: [ dropUser() ]']);
         }
@@ -91,6 +109,12 @@ const handleMessage = function (ctx, user, msg) {
         let chanName = obj || randName();
         sendMsg(ctx, user, [seq, 'JACK', chanName]);
         let chan = ctx.channels[chanName] = ctx.channels[chanName] || [];
+
+        // prevent removal of the channel if there is a pending timeout
+        if (ctx.config.removeChannels && ctx.timeouts[chanName]) {
+            clearTimeout(ctx.timeouts[chanName]);
+        }
+
         chan.id = chanName;
         if (USE_HISTORY_KEEPER) {
             sendMsg(ctx, user, [0, HISTORY_KEEPER_ID, 'JOIN', chanName]);
@@ -153,11 +177,19 @@ const handleMessage = function (ctx, user, msg) {
     }
 };
 
-let run = module.exports.run = function (storage, socketServer) {
+let run = module.exports.run = function (storage, socketServer, config) {
+    /*  Channel removal timeout defaults to 60000ms (one minute) */
+    config.channelRemovalTimeout =
+        typeof(config.channelRemovalTimeout) === 'number'?
+            config.channelRemovalTimeout:
+            60000;
+
     let ctx = {
         users: {},
         channels: {},
-        store: (USE_FILE_BACKUP_STORAGE) ? LogStore.create('messages.log', storage) : storage
+        timeouts: {},
+        store: (USE_FILE_BACKUP_STORAGE) ? LogStore.create('messages.log', storage) : storage,
+        config: config
     };
     setInterval(function () {
         Object.keys(ctx.users).forEach(function (userId) {
@@ -183,7 +215,7 @@ let run = module.exports.run = function (storage, socketServer) {
         ctx.users[user.id] = user;
         sendMsg(ctx, user, [0, '', 'IDENT', user.id]);
         socket.on('message', function(message) {
-            if (LOG_MESSAGES) { console.log('>'+message); }
+            if (ctx.config.logToStdout) { console.log('>'+message); }
             try {
                 handleMessage(ctx, user, message);
             } catch (e) {

config.js.dist

@@ -12,6 +12,20 @@ module.exports = {
     // the port used for websockets
     websocketPort: 3000,
 
+    /*  Cryptpad can log activity to stdout
+     *  This may be useful for debugging
+     */
+    logToStdout: false,
+
+    /*  Cryptpad can be configured to remove channels some number of ms
+        after the last remaining client has disconnected.
+
+        Default behaviour is to keep channels forever.
+        If you enable channel removal, the default removal time is one minute
+    */
+    removeChannels: false,
+    channelRemovalTimeout: 60000,
+
     // You now have a choice of storage engines
 
     /* amnesiadb only exists in memory.

server.js

@@ -81,6 +81,6 @@ if (config.websocketPort !== config.httpPort) {
 var wsSrv = new WebSocketServer(wsConfig);
 Storage.create(config, function (store) {
     console.log('DB connected');
-    NetfluxSrv.run(store, wsSrv);
+    NetfluxSrv.run(store, wsSrv, config);
     WebRTCSrv.run(wsSrv);
 });

storage/LogStore.js

@@ -8,11 +8,12 @@ var create = module.exports.create = function(filePath, backingStore) {
 
     var file = Fs.createWriteStream(filePath, {flags: 'a+'});
 
-    return {
-        message: function(channel, msg, callback) {
-            message(file, msg);
-            backingStore.message(channel, msg, callback);
-        },
-        getMessages: backingStore.getMessages
+    var originalMessageFunction = backingStore.message;
+
+    backingStore.message = function(channel, msg, callback) {
+        message(file, msg);
+        originalMessageFunction(channel, msg, callback);
     };
+
+    return backingStore;
 };

storage/README.md

@@ -42,6 +42,12 @@ This function accepts the name of the channel in which the user is interested, t
 It is only implemented within the leveldb adaptor, making our latest code incompatible with the other back ends.
 While we migrate to our new Netflux API, only the leveldb adaptor will be supported.
 
+## removeChannel(channelName, callback)
+
+This method is called (optionally, see config.js.dist for more info) some amount of time after the last client in a channel disconnects.
+
+It should remove any history of that channel, and execute a callback which takes an error message as an argument.
+
 ## Documenting your adaptor
 
 Naturally, you should comment your code well before making a PR.

storage/amnesia.js

@@ -19,6 +19,10 @@ module.exports.create = function(conf, cb){
     var db=[],
         index=0;
 
+    if (conf.removeChannels) {
+        console.log("Server is set to remove channels %sms after the last remaining client leaves.", conf.channelRemovalTimeout);
+    }
+
     cb({
         message: function(channelName, content, cb){
             var val = {
@@ -41,5 +45,12 @@ module.exports.create = function(conf, cb){
             });
             if (cb) { cb(); }
         },
+        removeChannel: function (channelName, cb) {
+            var err = false;
+            db = db.filter(function (msg) {
+                return msg.chan !== channelName;
+            });
+            cb(err);
+        },
     });
 };