broker.subscribeParticipantChange(): Add callback

Author: jonnor

src/amqp.coffee

@@ -203,7 +203,12 @@ class MessageBroker extends Client
     return callback null
 
   # Participant registration
-  subscribeParticipantChange: (handler) ->
+  subscribeParticipantChange: (handler, callback) ->
+    defaultCallback = (err) ->
+      if err
+        console.err "Error in msgflo.amqp.subscribeParticipantChange, and no callback added", err
+    callback = defaultCallback if not callback
+
     deserialize = (message) =>
       debug 'receive on fbp', message.fields.deliveryTag
       data = null
@@ -218,6 +223,7 @@ class MessageBroker extends Client
 
     @channel.assertQueue 'fbp'
     @channel.consume 'fbp', deserialize
+    return callback null
 
 exports.Client = Client
 exports.MessageBroker = MessageBroker

src/direct.coffee

@@ -121,9 +121,16 @@ class MessageBroker extends interfaces.MessageBroker
   nackMessage: (message) ->
     return
 
-  subscribeParticipantChange: (handler) ->
+  subscribeParticipantChange: (handler, callback) ->
+    defaultCallback = (err) ->
+      if err
+        console.err "Error in msgflo.direct.subscribeParticipantChange, and no callback added", err
+    callback = defaultCallback if not callback
+
     @createQueue '', 'fbp', (err) =>
-      @subscribeToQueue 'fbp', handler, () ->
+      return callback err if err
+      @subscribeToQueue 'fbp', handler, (err) ->
+        return callback err
 
 exports.MessageBroker = MessageBroker
 exports.Client = Client

src/interfaces.coffee

@@ -73,7 +73,7 @@ class MessageBroker extends MessagingSystem
     throw new Error 'Not Implemented'
 
   # Participant registration
-  subscribeParticipantChange: (handler) ->
+  subscribeParticipantChange: (handler, callback) ->
     throw new Error 'Not Implemented'
 
 exports.MessageBroker = MessageBroker

src/mqtt.coffee

@@ -125,9 +125,16 @@ class MessageBroker extends Client
     routing.binderMixin this
 
   # Participant registration
-  subscribeParticipantChange: (handler) ->
+  subscribeParticipantChange: (handler, callback) ->
+    defaultCallback = (err) ->
+      if err
+        console.err "Error in msgflo.mqtt.subscribeParticipantChange, and no callback added", err
+    callback = defaultCallback if not callback
+
     @createQueue '', 'fbp', (err) =>
-      @subscribeToQueue 'fbp', handler, () ->
+      return callback err if err
+      @subscribeToQueue 'fbp', handler, (err) ->
+        return callback err
 
 exports.Client = Client
 exports.MessageBroker = MessageBroker