Merge pull request #24 from msgflo/mqtt-subscribe

Support subscribing to data on MQTT

Author: jonnor

Gruntfile.coffee

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

spec/01transport.coffee

@@ -99,6 +99,7 @@ transportTests = (type) ->
   broker = null
 
   describeIfRoundRobinSupport = if type == 'AMQP' then describe else describe.skip
+  describeIfSubscribeSupport = if type != 'AMQP' then describe else describe.skip
 
   beforeEach (done) ->
     broker = transport.getBroker address
@@ -149,8 +150,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 +174,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,8 +201,8 @@ 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'
+              clients.sender.sendTo 'outqueue', outQueue, payload, (err) ->
+                chai.expect(err).to.be.a 'null'
 
 
   describe 'multiple outqueues bound to one inqueue', ->
@@ -363,6 +364,100 @@ transportTests = (type) ->
       it 'only NACKed message is sent to deadletter', ->
         chai.expect(received.deadletter).to.eql [ { foo: 'nack'} ]
 
+  describeIfSubscribeSupport '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
+
+  describeIfSubscribeSupport '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

@@ -190,12 +190,18 @@ class MessageBroker extends Client
 
     else
       return callback new Error 'Unsupported binding type: '+binding.type
-  removeBinding: (binding, callback) ->
-    # FIXME: implement
+
+  removeBinding: (binding, callback) -> # FIXME: implement
     return callback null
-  listBindings: (from, callback) ->
+  listBindings: (from, callback) -> # FIXME: implement
     return callback null, []
-    
+
+  # Data subscriptions
+  subscribeData: (binding, datahandler, callback) -> # TODO: implement
+    return callback null
+  unsubscribeData: (binding, datahandler, callback) -> # TODO: implement
+    return callback null
+
   # Participant registration
   subscribeParticipantChange: (handler) ->
     deserialize = (message) =>

src/interfaces.coffee

@@ -66,6 +66,12 @@ 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'
+
   # Participant registration
   subscribeParticipantChange: (handler) ->
     throw new Error 'Not Implemented'

src/routing.coffee

@@ -30,21 +30,32 @@ bindingId = (f, t) ->
 class Binder
   constructor: (@transport) ->
     @bindings = {}
+    @subscriptions = {}
 
   addBinding: (binding, callback) ->
     from = binding.src
     to = binding.tgt
     # TODO: handle non-pubsub types
     id = bindingId from, to
     debug 'Binder.addBinding', binding.type, id
-    return callback null if @bindings[id] or from == to
+    return callback null if @bindings[id] # already exists, avoid duplicate
 
     handler = (msg) =>
       binding = @bindings[id]
       return if not binding?.enabled
       debug 'edge message', from, to, msg
-      @transport.sendTo 'outqueue', to, msg.data, (err) ->
-        throw err if err
+
+      subscription = @subscriptions[id]
+      if subscription
+        for subCallback in subscription.handlers
+          subCallback(subscription.binding, msg.data)
+
+      if from != to
+        @transport.sendTo 'outqueue', to, msg.data, (err) ->
+          throw err if err
+      else
+        # same topic/queue, data should appear without our forwarding
+
     @transport.subscribeToQueue from, handler, (err) =>
       return callback err if err
       @bindings[id] =
@@ -68,6 +79,18 @@ class Binder
     debug 'Binder.listBindings'
     return callback null, []
 
+  subscribeData: (binding, datahandler, callback) ->
+    id = bindingId binding.src, binding.tgt
+    @subscriptions[id] = { handlers: [], binding: binding } if not @subscriptions[id]
+    @subscriptions[id].handlers.push datahandler
+    return callback null
+  unsubscribeData: (binding, datahandler, callback) ->
+    id = bindingId binding.src, binding.tgt
+    subscription = @subscriptions[id]
+    handlerIndex = subscription.handlers.indexOf datahandler
+    return callback new Error "Subscription was not found" if handlerIndex == -1
+    subscription.handlers = subscription.handlers.splice(handlerIndex, 1)
+    return callback null
 
 exports.Binder = Binder
 exports.binderMixin = (transport) ->
@@ -76,4 +99,6 @@ exports.binderMixin = (transport) ->
   transport.addBinding = b.addBinding.bind b
   transport.removeBinding = b.removeBinding.bind b
   transport.listBindings = b.listBindings.bind b
+  transport.subscribeData = b.subscribeData.bind b
+  transport.unsubscribeData = b.unsubscribeData.bind b