Add destination mw (#148)

* feat: add addDestinationMiddleware function

WIP

* feat: fix bug whereby changes made by integration mw were not persisted

This bug meant that this code did not send an event with a userId=bar
to segment.io. This change will result in this code sending an event
with userId=bar to segment.io:

f=function(payload, integration, next) {
	payload.obj.userId = "bar";next(payload);
};
analytics.addIntegrationMiddleware(f);
analytics.identify();

* feat: correct doc

* feat: update addDestinationMiddleware docs

* feat: update dest mw signature to be consistent with source mw

* chore: fix syntax error

* chore: fix failing tests

* chore: fix failing test by enabling integration

Co-authored-by: Matthew Ault <[email protected]>

Author: aultimus

lib/analytics.js

@@ -14,6 +14,8 @@ var Identify = require('segmentio-facade').Identify;
 var SourceMiddlewareChain = require('./middleware').SourceMiddlewareChain;
 var IntegrationMiddlewareChain = require('./middleware')
   .IntegrationMiddlewareChain;
+var DestinationMiddlewareChain = require('./middleware')
+  .DestinationMiddlewareChain;
 var Page = require('segmentio-facade').Page;
 var Track = require('segmentio-facade').Track;
 var bindAll = require('bind-all');
@@ -50,6 +52,7 @@ function Analytics() {
   this.Integrations = {};
   this._sourceMiddlewares = new SourceMiddlewareChain();
   this._integrationMiddlewares = new IntegrationMiddlewareChain();
+  this._destinationMiddlewares = {};
   this._integrations = {};
   this._readied = false;
   this._timeout = 300;
@@ -111,6 +114,7 @@ Analytics.prototype.addSourceMiddleware = function(middleware) {
 
 /**
  * Define a new `IntegrationMiddleware`
+ * DEPRECATED
  *
  * @param {Function} Middleware
  * @return {Analytics}
@@ -121,6 +125,32 @@ Analytics.prototype.addIntegrationMiddleware = function(middleware) {
   return this;
 };
 
+/**
+ * Define a new `DestinationMiddleware`
+ * Destination Middleware is chained after integration middleware
+ *
+ * @param {String} integrationName
+ * @param {Array} Middlewares
+ * @return {Analytics}
+ */
+
+Analytics.prototype.addDestinationMiddleware = function(
+  integrationName,
+  middlewares
+) {
+  var self = this;
+  middlewares.forEach(function(middleware) {
+    if (!self._destinationMiddlewares[integrationName]) {
+      self._destinationMiddlewares[
+        integrationName
+      ] = new DestinationMiddlewareChain();
+    }
+
+    self._destinationMiddlewares[integrationName].add(middleware);
+  });
+  return self;
+};
+
 /**
  * Initialize with the given integration `settings` and `options`.
  *
@@ -734,11 +764,6 @@ Analytics.prototype._invoke = function(method, facade) {
           result = new Facade(result);
         }
 
-        self.emit('invoke', result);
-        metrics.increment('analytics_js.invoke', {
-          method: method
-        });
-
         applyIntegrationMiddlewares(result);
       }
     );
@@ -791,12 +816,53 @@ Analytics.prototype._invoke = function(method, facade) {
                 result = new Facade(result);
               }
 
-              metrics.increment('analytics_js.integration.invoke', {
-                method: method,
-                integration_name: integration.name
-              });
-
-              integration.invoke.call(integration, method, result);
+              // apply destination middlewares
+              // Apply any integration middlewares that exist, then invoke the integration with the result.
+              if (self._destinationMiddlewares[integration.name]) {
+                self._destinationMiddlewares[integration.name].applyMiddlewares(
+                  facadeCopy,
+                  integration.name,
+                  function(result) {
+                    // A nullified payload should not be sent to an integration.
+                    if (result === null) {
+                      self.log(
+                        'Payload to destination "%s" was null and dropped by a middleware.',
+                        name
+                      );
+                      return;
+                    }
+
+                    // Check if the payload is still a Facade. If not, convert it to one.
+                    if (!(result instanceof Facade)) {
+                      result = new Facade(result);
+                    }
+
+                    self.emit('invoke', result);
+                    metrics.increment('analytics_js.invoke', {
+                      method: method
+                    });
+
+                    metrics.increment('analytics_js.integration.invoke', {
+                      method: method,
+                      integration_name: integration.name
+                    });
+
+                    integration.invoke.call(integration, method, result);
+                  }
+                );
+              } else {
+                self.emit('invoke', result);
+                metrics.increment('analytics_js.invoke', {
+                  method: method
+                });
+
+                metrics.increment('analytics_js.integration.invoke', {
+                  method: method,
+                  integration_name: integration.name
+                });
+
+                integration.invoke.call(integration, method, result);
+              }
             }
           );
         } catch (e) {

lib/middleware.js

@@ -34,6 +34,20 @@ module.exports.IntegrationMiddlewareChain = function IntegrationMiddlewareChain(
   };
 };
 
+module.exports.DestinationMiddlewareChain = function DestinationMiddlewareChain() {
+  var apply = middlewareChain(this);
+
+  this.applyMiddlewares = function(facade, integration, callback) {
+    return apply(
+      function(mw, payload, next) {
+        mw({ payload: payload, integration: integration, next: next });
+      },
+      facade,
+      callback
+    );
+  };
+};
+
 // Chain is essentially a linked list of middlewares to run in order.
 function middlewareChain(dest) {
   var middlewares = [];

test/analytics.test.js

@@ -495,7 +495,7 @@ describe('Analytics', function() {
     });
 
     it('should emit "invoke" with facade', function(done) {
-      var opts = { All: false };
+      var opts = { All: true };
       var identify = new Identify({ testVal: 'success', options: opts });
       analytics.on('invoke', function(msg) {
         assert(msg instanceof Facade);
@@ -2058,6 +2058,51 @@ describe('Analytics', function() {
     });
   });
 
+  describe('#addDestinationMiddleware', function() {
+    it('should have a defined _integrationMiddlewares property', function() {
+      assert(analytics._destinationMiddlewares !== undefined);
+    });
+
+    it('should allow users to add a valid Middleware', function() {
+      try {
+        analytics.addDestinationMiddleware('foo', [function() {}]);
+      } catch (e) {
+        // This assert should not run.
+        assert(false, 'error was incorrectly thrown!');
+      }
+    });
+
+    it('should throw an error if the selected Middleware is not a function', function() {
+      try {
+        analytics.addDestinationMiddleware('foo', [7]);
+
+        // This assert should not run.
+        assert(false, 'error was not thrown!');
+      } catch (e) {
+        assert(
+          e.message === 'attempted to add non-function middleware',
+          'wrong error return'
+        );
+      }
+    });
+
+    it('should not throw an error if AJS has already initialized', function() {
+      analytics.init();
+      try {
+        analytics.addDestinationMiddleware('foo', [function() {}]);
+      } catch (e) {
+        // This assert should not run.
+        assert(false, 'error was thrown!');
+      }
+    });
+
+    it('should return the analytics object', function() {
+      assert(
+        analytics === analytics.addDestinationMiddleware('foo', [function() {}])
+      );
+    });
+  });
+
   describe('#addSourceMiddleware', function() {
     it('should have a defined _sourceMiddlewares property', function() {
       assert(analytics._sourceMiddlewares !== undefined);