Allow replacing entire req.data property in all realtime middleware

Author: jondubois

scserver.js

@@ -535,11 +535,11 @@ SCServer.prototype._isPrivateTransmittedEvent = function (event) {
   return !!event && event.indexOf('#') == 0;
 };
 
-SCServer.prototype.verifyInboundEvent = function (socket, event, data, cb) {
+SCServer.prototype.verifyInboundEvent = function (socket, eventName, eventData, cb) {
   var request = {
     socket: socket,
-    event: event,
-    data: data
+    event: eventName,
+    data: eventData
   };
 
   var token = socket.getAuthToken();
@@ -578,15 +578,15 @@ SCServer.prototype._passThroughMiddleware = function (options, cb) {
 
   if (this._isPrivateTransmittedEvent(event)) {
     if (event == this._subscribeEvent) {
-      var data = options.data || {};
-      request.channel = data.channel;
-      request.waitForAuth = data.waitForAuth;
-      request.data = data.data;
+      var eventData = options.data || {};
+      request.channel = eventData.channel;
+      request.waitForAuth = eventData.waitForAuth;
+      request.data = eventData.data;
 
       if (request.waitForAuth && request.authTokenExpiredError) {
         // If the channel has the waitForAuth flag set, then we will handle the expiry quietly
         // and we won't pass this request through the subscribe middleware.
-        cb(request.authTokenExpiredError);
+        cb(request.authTokenExpiredError, eventData);
       } else {
         async.applyEachSeries(this._middleware[this.MIDDLEWARE_SUBSCRIBE], request,
           function (err) {
@@ -601,37 +601,43 @@ SCServer.prototype._passThroughMiddleware = function (options, cb) {
                   self.emit('warning', err);
                 }
               }
-              cb(err);
+              if (request.data !== undefined) {
+                eventData.data = request.data;
+              }
+              cb(err, eventData);
             }
           }
         );
       }
     } else if (event == this._publishEvent) {
       if (this.allowClientPublish) {
-        var data = options.data || {};
-        request.channel = data.channel;
-        request.data = data.data;
+        var eventData = options.data || {};
+        request.channel = eventData.channel;
+        request.data = eventData.data;
 
         async.applyEachSeries(this._middleware[this.MIDDLEWARE_PUBLISH_IN], request,
           function (err) {
             if (callbackInvoked) {
               self.emit('warning', new InvalidActionError('Callback for ' + self.MIDDLEWARE_PUBLISH_IN + ' middleware was already invoked'));
             } else {
               callbackInvoked = true;
+              if (request.data !== undefined) {
+                eventData.data = request.data;
+              }
               if (err) {
                 if (err === true) {
                   err = new SilentMiddlewareBlockedError('Action was silently blocked by ' + self.MIDDLEWARE_PUBLISH_IN + ' middleware', self.MIDDLEWARE_PUBLISH_IN);
                 } else if (self.middlewareEmitWarnings) {
                   self.emit('warning', err);
                 }
-                cb(err);
+                cb(err, eventData);
               } else {
                 self.exchange.publish(request.channel, request.data, function (err) {
                   if (err) {
                     err = new BrokerError(err);
                     self.emit('warning', err);
                   }
-                  cb(err);
+                  cb(err, eventData);
                 });
               }
             }
@@ -640,11 +646,11 @@ SCServer.prototype._passThroughMiddleware = function (options, cb) {
       } else {
         var noPublishError = new InvalidActionError('Client publish feature is disabled');
         self.emit('warning', noPublishError);
-        cb(noPublishError);
+        cb(noPublishError, options.data);
       }
     } else {
       // Do not allow blocking other reserved events or it could interfere with SC behaviour
-      cb();
+      cb(null, options.data);
     }
   } else {
     request.event = event;
@@ -663,48 +669,51 @@ SCServer.prototype._passThroughMiddleware = function (options, cb) {
               self.emit('warning', err);
             }
           }
-          cb(err);
+          cb(err, request.data);
         }
       }
     );
   }
 };
 
-SCServer.prototype.verifyOutboundEvent = function (socket, event, data, options, cb) {
+SCServer.prototype.verifyOutboundEvent = function (socket, eventName, eventData, options, cb) {
   var self = this;
 
   var callbackInvoked = false;
 
-  if (event == this._publishEvent) {
+  if (eventName == this._publishEvent) {
     var request = {
       socket: socket,
-      channel: data.channel,
-      data: data.data
+      channel: eventData.channel,
+      data: eventData.data
     };
     async.applyEachSeries(this._middleware[this.MIDDLEWARE_PUBLISH_OUT], request,
       function (err) {
         if (callbackInvoked) {
           self.emit('warning', new InvalidActionError('Callback for ' + self.MIDDLEWARE_PUBLISH_OUT + ' middleware was already invoked'));
         } else {
           callbackInvoked = true;
+          if (request.data !== undefined) {
+            eventData.data = request.data;
+          }
           if (err) {
             if (err === true) {
               err = new SilentMiddlewareBlockedError('Action was silently blocked by ' + self.MIDDLEWARE_PUBLISH_OUT + ' middleware', self.MIDDLEWARE_PUBLISH_OUT);
             } else if (self.middlewareEmitWarnings) {
               self.emit('warning', err);
             }
-            cb(err);
+            cb(err, eventData);
           } else {
             if (options && request.useCache) {
               options.useCache = true;
             }
-            cb();
+            cb(null, eventData);
           }
         }
       }
     );
   } else {
-    cb();
+    cb(null, eventData);
   }
 };
 

scsocket.js

@@ -98,19 +98,20 @@ var SCSocket = function (id, server, socket) {
 
         if (self._localEvents[eventName] == null) {
           var response = new Response(self, obj.cid);
-          self.server.verifyInboundEvent(self, eventName, obj.data, function (err) {
+          self.server.verifyInboundEvent(self, eventName, obj.data, function (err, newData) {
             if (err) {
               response.error(err);
             } else {
+              var eventData = newData;
               if (eventName == '#disconnect') {
-                var eventData = obj.data || {};
-                self._onSCClose(eventData.code, eventData.data);
+                var disconnectData = eventData || {};
+                self._onSCClose(disconnectData.code, disconnectData.data);
               } else {
                 if (self._autoAckEvents[eventName]) {
                   response.end();
-                  SCEmitter.prototype.emit.call(self, eventName, obj.data);
+                  SCEmitter.prototype.emit.call(self, eventName, eventData);
                 } else {
-                  SCEmitter.prototype.emit.call(self, eventName, obj.data, response.callback.bind(response));
+                  SCEmitter.prototype.emit.call(self, eventName, eventData, response.callback.bind(response));
                 }
               }
             }
@@ -249,12 +250,12 @@ SCSocket.prototype.emit = function (event, data, callback, options) {
   var self = this;
 
   if (this._localEvents[event] == null) {
-    this.server.verifyOutboundEvent(this, event, data, options, function (err) {
+    this.server.verifyOutboundEvent(this, event, data, options, function (err, newData) {
       var eventObject = {
         event: event
       };
-      if (data !== undefined) {
-        eventObject.data = data;
+      if (newData !== undefined) {
+        eventObject.data = newData;
       }
       eventObject.cid = self._nextCallId();