Merge pull request #1740 from ether/feature/add-cs-queue

add a changeset queue

src/node/handler/PadMessageHandler.js

@@ -35,6 +35,7 @@ var messageLogger = log4js.getLogger("message");
 var accessLogger = log4js.getLogger("access");
 var _ = require('underscore');
 var hooks = require("ep_etherpad-lite/static/js/pluginfw/hooks.js");
+var channels = require("channels");
 
 /**
  * A associative array that saves informations about a session
@@ -48,6 +49,11 @@ var hooks = require("ep_etherpad-lite/static/js/pluginfw/hooks.js");
  */
 var sessioninfos = {};
 
+/**
+ * A changeset queue per pad that is processed by handleUserChanges()
+ */
+var padChannels = new channels.channels(handleUserChanges);
+
 /**
  * Saves the Socket class we need to send and recieve data from the client
  */
@@ -176,7 +182,7 @@ exports.handleMessage = function(client, message)
       if (sessioninfos[client.id].readonly) {
         messageLogger.warn("Dropped message, COLLABROOM for readonly pad");
       } else if (message.data.type == "USER_CHANGES") {
-        handleUserChanges(client, message);
+        padChannels.emit(message.padId, {client: client, message: message});// add to pad queue
       } else if (message.data.type == "USERINFO_UPDATE") {
         handleUserInfoUpdate(client, message);
       } else if (message.data.type == "CHAT_MESSAGE") {
@@ -522,23 +528,26 @@ function handleUserInfoUpdate(client, message)
  * @param client the client that send this message
  * @param message the message from the client
  */
-function handleUserChanges(client, message)
+function handleUserChanges(data, cb)
 {
+  var client = data.client
+    , message = data.message
+
   // Make sure all required fields are present
   if(message.data.baseRev == null)
   {
     messageLogger.warn("Dropped message, USER_CHANGES Message has no baseRev!");
-    return;
+    return cb();
   }
   if(message.data.apool == null)
   {
     messageLogger.warn("Dropped message, USER_CHANGES Message has no apool!");
-    return;
+    return cb();
   }
   if(message.data.changeset == null)
   {
     messageLogger.warn("Dropped message, USER_CHANGES Message has no changeset!");
-    return;
+    return cb();
   }
  
   //get all Vars we need
@@ -679,6 +688,7 @@ function handleUserChanges(client, message)
     }
   ], function(err)
   {
+    cb();
     ERR(err);
   });
 }

src/node/handler/SocketIORouter.js

@@ -48,88 +48,56 @@ exports.addComponent = function(moduleName, module)
 /**
  * sets the socket.io and adds event functions for routing
  */
-exports.setSocketIO = function(_socket)
-{
+exports.setSocketIO = function(_socket) {
   //save this socket internaly
   socket = _socket;
   
-  socket.sockets.on('connection', function(client)
-  {
+  socket.sockets.on('connection', function(client) {
     client.set('remoteAddress', client.handshake.address.address);
     var clientAuthorized = false;
     
     //wrap the original send function to log the messages
     client._send = client.send;
-    client.send = function(message)
-    {
+    client.send = function(message) {
       messageLogger.debug("to " + client.id + ": " + stringifyWithoutPassword(message));
       client._send(message);
     }
   
     //tell all components about this connect
-    for(var i in components)
-    {
+    for(var i in components) {
       components[i].handleConnect(client);
     } 
 
-    //try to handle the message of this client
-    function handleMessage(message)
-    {
-      if(message.component && components[message.component])
-      {
-        //check if component is registered in the components array        
-        if(components[message.component])
-        {
-          messageLogger.debug("from " + client.id + ": " + stringifyWithoutPassword(message));
-          components[message.component].handleMessage(client, message);
-        }
-      }
-      else
-      {
-        messageLogger.error("Can't route the message:" + stringifyWithoutPassword(message));
-      }
-    }  
-      
     client.on('message', function(message)
     {
-      if(message.protocolVersion && message.protocolVersion != 2)
-      {
+      if(message.protocolVersion && message.protocolVersion != 2) {
         messageLogger.warn("Protocolversion header is not correct:" + stringifyWithoutPassword(message));
         return;
       }
 
       //client is authorized, everything ok
-      if(clientAuthorized)
-      {
-        handleMessage(message);
-      }
-      //try to authorize the client
-      else
-      {
+      if(clientAuthorized) {
+        handleMessage(client, message);
+      } else { //try to authorize the client
+        if(message.padId !== undefined && message.sessionID !== undefined && message.token !== undefined && message.password !== undefined) {
           //this message has everything to try an authorization
-        if(message.padId !== undefined && message.sessionID !== undefined && message.token !== undefined && message.password !== undefined)
-        {
-          securityManager.checkAccess (message.padId, message.sessionID, message.token, message.password, function(err, statusObject)
-          {
+          securityManager.checkAccess (message.padId, message.sessionID, message.token, message.password,
+            function(err, statusObject) {
               ERR(err);
 
               //access was granted, mark the client as authorized and handle the message
-            if(statusObject.accessStatus == "grant")
-            {
+              if(statusObject.accessStatus == "grant") {
                 clientAuthorized = true;
-              handleMessage(message);
+                handleMessage(client, message);
               }
               //no access, send the client a message that tell him why
-            else
-            {
+              else {
                 messageLogger.warn("Authentication try failed:" + stringifyWithoutPassword(message));
                 client.json.send({accessStatus: statusObject.accessStatus});
               }
-          });
             }
-        //drop message
-        else
-        {
+          );
+        } else { //drop message
           messageLogger.warn("Dropped message cause of bad permissions:" + stringifyWithoutPassword(message));
         }
       }
@@ -146,6 +114,21 @@ exports.setSocketIO = function(_socket)
   });
 }
 
+//try to handle the message of this client
+function handleMessage(client, message)
+{
+
+  if(message.component && components[message.component]) {
+    //check if component is registered in the components array
+    if(components[message.component]) {
+      messageLogger.debug("from " + client.id + ": " + stringifyWithoutPassword(message));
+      components[message.component].handleMessage(client, message);
+    }
+  } else {
+    messageLogger.error("Can't route the message:" + stringifyWithoutPassword(message));
+  }
+} 
+
 //returns a stringified representation of a message, removes the password
 //this ensures there are no passwords in the log
 function stringifyWithoutPassword(message)

src/package.json

@@ -37,7 +37,8 @@
                       "underscore"              : "1.3.1",
                       "unorm"                   : "1.0.0",
                       "languages4translatewiki" : "0.1.3",
-                      "swagger-node-express"    : "1.2.3"
+                      "swagger-node-express"    : "1.2.3",
+                      "channels"                : "0.0.x"
                      },
   "bin":             { "etherpad-lite": "./node/server.js" },
   "devDependencies": {