Merge branch 'master' into greenkeeper/mqtt-2.3.1

Author: jonnor

.travis.yml

@@ -24,7 +24,7 @@ deploy:
   email: [email protected]
   skip_cleanup: true
   api_key:
-    secure: K+uFHsh/EgT6zKDdXJV+770WdXzuBaiLQkN1EI/JU5AfjOqw8cWI2ycbH9LWlOCcE29l/cVFL+ycx/1x69IwmAWk61TlwPXqT1KiDBn8lxEN2dPub/0Cr7iHP/XV60daao2TJS+KS44wS6MIMcwBNkbeCZXsOKQZjFR/qjlAM5kCCQO+cZBWC+m2k3w0KHRO+6AdAOke205iGrI0/ZHLNU96f0jhMtwy3oM2DQEwY6aerAVx6cBes9Cdqzn86qgOohP5XXVpllTzeBTt6WjIQsQB1PXFReYfN0Tfcsvg2bcOUMhc5Exj1p40/Q9gSNKZon/4kRQFKvut1+189qMga6JKA6wdYO8xltVgRgwGc6pyeXMrJHZ/vzxyOLCYTE/W15Kg61dXrzBi4Ke+1Bm7DGuYFvAanSgEboseNFCkKSNirDO2a2Xv2ldRlWQpgEOADhAczGSlmUTz2/UqMEwrvc1M+DyRnGD2zyDV4My23jGAKnI/vqFke3PvLjGa+GpzcvEYntEq2iCIDSaoaJH47d1T9Rj2/g5Vl9hw6bfkLZIFk46fUfU4KzZJHmxs3x7s0vijNY2DAT3TrLhIw20gMbQJBiAW1BCojgZ2B+OB/ChYQcxYJ2IQEJxsLBXFrlSC3tfXCzWgf7yjrwYsP3Hbj+AbP+74E7uwLh/eSDfrpCY=
+    secure: lovNd++XUGKSGeO0ze6rI87Y/RCdEvOlD1Z4FzVaCbjD6meNiOygf4SrHdVMUa8Xe0kCBUc9MRVqKZEvvgJ/yLgv+LwwU6+jhecGMWmovVh0b7q3il7ixzdvR4Lyyl4YxNu2AJMFn6G/YzniNm5EwaIss5vkoUAijmfys5FQFMsj/IQp08u2PRusfSua0Nnx5FtMDS17k2bqheCE7rTMlHswHuHCrE0gabBZoMgLK1K3DGvROvvQzbMlMYrLLbT8Xdm08opqzIUYTHqo7fIZOFqPieuw9gPb6T/M3QY9k0UIflF/rh9hoCAdJRa/BT8pxwXmE2V0PxKGMCTzphfKQ1mfpqQCTM4jNPDSRhadkAe1JTas9PeP91IPSECcamN9soBW/3fUe4ji3UGpaJHmU97Owd3BBzNQl2PMoHUBmyI4C0QkwukjgyStky/oEFaL6uh/scA2Hrin1jsOv2sXEltjkogbRWspGv9q+1DfJkHeiYqpp44nU4Kb2dXVNFqYSVxNUNewRrY3hsH8K1cBK4x61jBOY5ItUNQ2a64GhZ7WyHr2gIPxwT33HD+4ksRnEknl6KU9qc+6kwdi7fERQ8nEyDUOv4ZWtvvoHXiwZzu7m7Xwhtj4RdJ2IEUniJlmHq7M/yr8FPGCP+AXQzBQq2BUPt3OUEWAYrlk0vfjFA4=
   on:
     tags: true
     repo: msgflo/msgflo-nodejs

CHANGES.md

@@ -1,4 +1,22 @@
 
+# msgflo-nodejs 0.10.0, 15.03.2017
+
+* Transport: Add interface for listing existing datasubscriptions.
+`MessageBroker.listSubscriptions()`
+
+# msgflo-nodejs 0.9.0, 04.02.2017
+
+* Transport: Add new interface for subscribing to data changes on a binding.
+See MessageBroker `subscribeData()` and `unsubscribeData()`.
+* MQTT: Support `subscribeData()`
+* MQTT: Fix falsy messages not being forwarded
+* MQTT: Fix addBinding() not working if a removeBinding() had been done before
+* No support for subscribeData on AMQP yet
+
+# msgflo-nodejs 0.8.2, 01.02.2017
+
+* MsgFlo discovery message now sent periodically. Defaults to couple of times per minute.
+
 # msgflo-nodejs 0.7.2, 09.10.2016
 
 * MQTT: Fix compatibility with newer versions of library (> 1.4.x)

Gruntfile.coffee

@@ -18,6 +18,7 @@ module.exports = ->
         src: ['spec/*.coffee']
         options:
           reporter: 'spec'
+          grep: process.env.TESTS
 
     # Coding standards
     coffeelint:

package.json

@@ -1,7 +1,7 @@
 {
   "name": "msgflo-nodejs",
   "description": "Node.js participant support for MsgFlo",
-  "version": "0.7.2",
+  "version": "0.10.5",
   "author": {
     "name": "Jon Nordby",
     "email": "[email protected]"
@@ -17,7 +17,7 @@
     "commander": "^2.6.0",
     "debug": "^2.1.2",
     "fbp": "^1.1.2",
-    "newrelic": "1.36.0",
+    "newrelic": "^1.39.0",
     "uuid": "^3.0.1"
   },
   "devDependencies": {
@@ -29,8 +29,8 @@
     "grunt-contrib-coffee": "^1.0.0",
     "grunt-mocha-test": "^0.13.2",
     "grunt-shell-spawn": "^0.3.1",
-    "mocha": "~3.2.0",
-    "mqtt": "^2.3.1"
+    "mocha": "~3.4.0",
+    "mqtt": "^2.7.1"
   },
   "keywords": [],
   "bin": {

spec/01transport.coffee

@@ -149,8 +149,8 @@ transportTests = (type) ->
 
           clients.receiver.subscribeToQueue sharedQueue, onReceive, (err) ->
             chai.expect(err).to.be.a 'null'
-          clients.sender.sendTo 'outqueue', sharedQueue, payload, (err) ->
-            chai.expect(err).to.be.a 'null'
+            clients.sender.sendTo 'outqueue', sharedQueue, payload, (err) ->
+              chai.expect(err).to.be.a 'null'
 
 
   describe 'inqueue==outqueue with binding', ->
@@ -173,8 +173,8 @@ transportTests = (type) ->
 
             clients.receiver.subscribeToQueue sharedQueue, onReceive, (err) ->
               chai.expect(err).to.be.a 'null'
-            clients.sender.sendTo 'outqueue', sharedQueue, payload, (err) ->
-              chai.expect(err).to.be.a 'null'
+              clients.sender.sendTo 'outqueue', sharedQueue, payload, (err) ->
+                chai.expect(err).to.be.a 'null'
 
 
   describe 'outqueue bound to inqueue', ->
@@ -200,6 +200,48 @@ transportTests = (type) ->
 
             clients.receiver.subscribeToQueue inQueue, onReceive, (err) ->
               chai.expect(err).to.be.a 'null'
+              clients.sender.sendTo 'outqueue', outQueue, payload, (err) ->
+                chai.expect(err).to.be.a 'null'
+
+  describe 'outqueue bound to inqueue then removed', ->
+    it 'sending to inqueue, show up on outqueue', (done) ->
+      payload = { foo: 'bar922' }
+      inQueue = 'inqueue922'
+      outQueue = 'outqueue922'
+      createConnectClients address, ['sender', 'receiver'], (err, clients) ->
+        createQueues [
+          [ clients.receiver, 'inqueue', inQueue ]
+          [ clients.sender, 'outqueue', outQueue ]
+        ], (err) ->
+          chai.expect(err).to.not.exist
+
+          binding = { type:'pubsub', src:outQueue, tgt:inQueue }
+          bindingRemoved = false
+
+          onReceive = (msg) ->
+            if bindingRemoved
+              done new Error "Received data on removed binding"
+              done = null
+              return
+
+            clients.receiver.ackMessage msg
+            chai.expect(msg).to.include.keys 'data'
+            chai.expect(msg.data).to.eql payload
+            bindingRemoved = true
+            broker.removeBinding binding, (err) ->
+              chai.expect(err).to.be.a 'null'
+              clients.sender.sendTo 'outqueue', outQueue, payload, (err) ->
+                chai.expect(err).to.be.a 'null'
+                setTimeout () ->
+                  done null if done
+                  done = null
+                  return
+                , 300
+
+          clients.receiver.subscribeToQueue inQueue, onReceive, (err) ->
+            chai.expect(err).to.be.a 'null'
+          broker.addBinding binding, (err) ->
+            chai.expect(err).to.be.a 'null'
             clients.sender.sendTo 'outqueue', outQueue, payload, (err) ->
               chai.expect(err).to.be.a 'null'
 
@@ -363,6 +405,100 @@ transportTests = (type) ->
       it 'only NACKed message is sent to deadletter', ->
         chai.expect(received.deadletter).to.eql [ { foo: 'nack'} ]
 
+  describe 'subscribing to bound topics', ->
+    sendQueue = 'sub-send-36'
+    receiveQueue = 'sub-receive-36'
+    binding = { type:'pubsub', src:sendQueue, tgt:receiveQueue }
+    connectionData = []
+    clients = null
+
+    # Should be a before, but the 'beforeEach' of higher scope are ran afterwards...
+    setup = (done) ->
+      createConnectClients address, ['sender', 'receiver'], (err, c) ->
+        clients = c
+        createQueues [
+          [ clients.receiver, 'inqueue', receiveQueue ]
+          [ clients.sender, 'outqueue', sendQueue ]
+        ], (err) ->
+          chai.expect(err).to.not.exist
+          broker.addBinding binding, (err) ->
+            chai.expect(err).to.be.a 'null'
+            return done null
+
+    it 'should provide data sent on connection', (done) ->
+      payloads =
+        one: { foo: 'sub-96' }
+        two: { bar: 'sub-97' }
+  
+      onData = (bind, data) ->
+        chai.expect(bind.src).to.equal binding.src
+        chai.expect(bind.tgt).to.equal binding.tgt
+        connectionData.push data
+        # wait until we've gotten two packets
+        if connectionData.length == 2
+          [one, two] = connectionData
+          chai.expect(one).to.eql payloads.one
+          chai.expect(two).to.eql payloads.two
+          return done null
+        else if connectionData.length > 2
+          return done new Error "Got more data than expected"
+
+      setup (err) ->
+        return done err if err
+        broker.subscribeData binding, onData, (err) ->
+          return done err if err
+          clients.sender.sendTo 'outqueue', sendQueue, payloads.one, (err) ->
+            return done err if err
+            clients.sender.sendTo 'outqueue', sendQueue, payloads.two, (err) ->
+              return done err if err
+
+  describe 'subscribing to binding with srcQueue==tgtQueue', ->
+    sendQueue = 'sub-shared-37'
+    receiveQueue = sendQueue
+    binding = { type:'pubsub', src:sendQueue, tgt:receiveQueue }
+    connectionData = []
+    clients = null
+
+    # Should be a before, but the 'beforeEach' of higher scope are ran afterwards...
+    setup = (done) ->
+      createConnectClients address, ['sender', 'receiver'], (err, c) ->
+        clients = c
+        createQueues [
+          [ clients.receiver, 'inqueue', receiveQueue ]
+          [ clients.sender, 'outqueue', sendQueue ]
+        ], (err) ->
+          chai.expect(err).to.not.exist
+          broker.addBinding binding, (err) ->
+            chai.expect(err).to.be.a 'null'
+            return done null
+
+    it 'should provide data sent on connection', (done) ->
+      payloads =
+        one: { foo: 'sub-106' }
+        two: { bar: 'sub-107' }
+
+      onData = (bind, data) ->
+        chai.expect(bind.src).to.equal binding.src
+        chai.expect(bind.tgt).to.equal binding.tgt
+        connectionData.push data
+        # wait until we've gotten two packets
+        if connectionData.length == 2
+          [one, two] = connectionData
+          chai.expect(one).to.eql payloads.one
+          chai.expect(two).to.eql payloads.two
+          return done null
+        else if connectionData.length > 2
+          return done new Error "Got more data than expected"
+
+      setup (err) ->
+        return done err if err
+        broker.subscribeData binding, onData, (err) ->
+          return done err if err
+          clients.sender.sendTo 'outqueue', sendQueue, payloads.one, (err) ->
+            return done err if err
+            clients.sender.sendTo 'outqueue', sendQueue, payloads.two, (err) ->
+              return done err if err
+
 describe 'Transport', ->
   Object.keys(transports).forEach (type) =>
     describe "#{type}", () ->

src/amqp.coffee

@@ -1,14 +1,15 @@
 
 debug = require('debug')('msgflo:amqp')
 async = require 'async'
+uuid = require 'uuid'
+
 interfaces = require './interfaces'
 
 try
   amqp = require 'amqplib/callback_api'
 catch e
   amqp = e
 
-
 class Client extends interfaces.MessagingClient
   constructor: (@address, @options={}) ->
     @connection = null
@@ -149,12 +150,55 @@ class Client extends interfaces.MessagingClient
     @channel.sendToQueue 'fbp', data
     return callback null
 
+
+dataSubscriptionQueueName = (id) ->
+  throw new Error("Missing id") if not id
+  return ".msgflo-broker-subscriptions-#{id}"
+
+bindingId = (b) ->
+  return "[#{b.src}]->[#{b.tgt}]"
+
 class MessageBroker extends Client
   constructor: (address, options) ->
     super address, options
+    @options.id = uuid.v4() if not @options.id
+    @subscriptions = {}
+
+  connect: (callback) ->
+    super (err) =>
+      return callback err if err
+      # create queue for data subscriptions
+      name = dataSubscriptionQueueName @options.id
+      options =
+        exclusive: true
+        durable: false
+        autoDelete: true
+      @channel.assertQueue name, options, (err) =>
+        return callback err if err
+        onSubscribedQueueData = (message) =>
+          exchange = message.fields.exchange
+          debug 'broker subscriber got message on exchange', exchange
+          matches = Object.keys(@subscriptions).filter (id) =>
+            sub = @subscriptions[id]
+            # XXX: how to account for which queue the message is for
+            # can we create some identifier when we subscribe?
+            return sub?.binding.src == exchange
+          for id in matches
+            sub = @subscriptions[id]
+            data = message.content
+            try
+              data = JSON.parse message.content.toString()
+            catch e
+              null
+            sub.handler sub.binding, data
+
+        subscribeOptions =
+          noAck: true
+        @channel.consume name, onSubscribedQueueData, subscribeOptions, (err) ->
+          debug 'broker created subscription queue', err
+          return callback err
 
   addBinding: (binding, callback) ->
-    # TODO: support roundrobin type
     debug 'Broker.addBinding', binding
     if binding.type == 'pubsub'
       @channel.bindQueue binding.tgt, binding.src, '', {}, callback
@@ -190,12 +234,49 @@ class MessageBroker extends Client
 
     else
       return callback new Error 'Unsupported binding type: '+binding.type
+
   removeBinding: (binding, callback) ->
-    # FIXME: implement
-    return callback null
-  listBindings: (from, callback) ->
+    debug 'Broker.removeBinding', binding
+    if binding.type == 'pubsub'
+      @channel.unbindQueue binding.tgt, binding.src, '', {}, callback
+    else if binding.type == 'roundrobin'
+      return callback new Error "removeBinding() not supported for type='roundrobin'" # TODO:
+    else
+      return callback new Error "Unsupported binding type: #{binding.type}"
+
+
+  listBindings: (from, callback) -> # FIXME: implement
+    # NOTE: probably need to use the RabbitMQ HTTP API for this
     return callback null, []
-    
+
+  # Data subscriptions
+  subscribeData: (binding, datahandler, callback) ->
+    exchange = binding.src
+    queue = dataSubscriptionQueueName @options.id
+    options =
+      autoDelete: true
+    @channel.bindQueue queue, exchange, '', options, (err) =>
+      return callback err if err
+      id = bindingId binding
+      @subscriptions[id] =
+        binding: binding
+        handler: datahandler
+      return callback null
+
+  unsubscribeData: (binding, datahandler, callback) ->
+    # TODO: also remove the subscription with broker
+    id = bindingId binding
+    delete @subscriptions[id]
+    return callback null
+
+  listSubscriptions: (callback) ->
+    # Is there a way to get this information through AMQP?
+    # Or do need to use RabbitMQ HTTP API?
+    subs = []
+    for id, sub of @subscriptions
+      subs.push sub.binding
+    return callback null, subs
+
   # Participant registration
   subscribeParticipantChange: (handler) ->
     deserialize = (message) =>

src/interfaces.coffee

@@ -37,7 +37,7 @@ class MessagingSystem
 class MessagingClient extends MessagingSystem
 
   # Participant registration
-  registerParticipant: (part) ->
+  registerParticipant: (part, callback) ->
     throw new Error 'Not Implemented'
 
 exports.MessagingClient = MessagingClient
@@ -66,6 +66,14 @@ class MessageBroker extends MessagingSystem
   listBindings: (callback) ->
     throw new Error 'Not Implemented'
 
+  # Subscribing to data on a binding
+  subscribeData: (binding, datahandler, callback) ->
+    throw new Error 'Not Implemented'
+  unsubscribeData: (binding, datahandler, callback) ->
+    throw new Error 'Not Implemented'
+  listSubscriptions: (callback) ->
+    throw new Error 'Not Implemented'
+
   # Participant registration
   subscribeParticipantChange: (handler) ->
     throw new Error 'Not Implemented'