Event Bus Scoping (#91)

* initial testing

* stashing changes

* testing

* pass in already parsed obj

lib/bus.js

@@ -3,7 +3,9 @@
 const bluebird = require('bluebird'),
   redis = require('redis'),
   streams = require('./streams'),
-  { BUS_NAMESPACE, BUS_TOPICS } = require('./constants');
+  { BUS_NAMESPACE, BUS_TOPICS } = require('./constants'),
+  PID = process.pid,
+  HOSTNAME = require('os').hostname();
 var client,
   log = require('./services/log').setup({file: __filename});
 
@@ -24,7 +26,16 @@ function connect() {
  */
 function disperseEvent(topic, payload) {
   try {
-    streams[topic].write(JSON.parse(payload));
+    let data = JSON.parse(payload),
+      scopedWrite = data.pid && data.hostname;
+
+    if (scopedWrite) { // We're dealing with amphora-redis-event-bus calls
+      if (data.pid === PID && HOSTNAME === data.hostname) {
+        streams[topic].write(data.msg);
+      }
+    } else {
+      streams[topic].write(data);
+    }
   } catch (e) {
     log('error', `Unable to send event ${topic} to bus: ${e.message}`, { payload });
   }

lib/bus.test.js

@@ -72,5 +72,31 @@ describe(filename, () => {
       expect(logMock).toHaveBeenCalled();
       expect(streams[batchEvent].write).not.toHaveBeenCalled();
     });
+
+    test('works with the payload of amphora-event-bus-redis', () => {
+      const msg = { foo: 'bar' },
+        eventBusData = {
+          hostname: require('os').hostname(),
+          pid: process.pid,
+          msg
+        };
+
+      streams[objEvent].write = jest.fn();
+      lib.disperseEvent(objEvent, JSON.stringify(eventBusData));
+      expect(streams[objEvent].write).toHaveBeenCalledWith(msg);
+    });
+
+    test('does not works with the payload of amphora-event-bus-redis if pid or host do not match', () => {
+      const msg = { foo: 'bar' },
+        eventBusData = {
+          hostname: require('os').hostname() + 1,
+          pid: process.pid + 1,
+          msg
+        };
+
+      streams[objEvent].write = jest.fn();
+      lib.disperseEvent(objEvent, JSON.stringify(eventBusData));
+      expect(streams[objEvent].write).not.toHaveBeenCalledWith(msg);
+    });
   });
 });