✨ Potential #30 implementation - returning promises in response callb…

…acks

lib/Response.js

@@ -26,20 +26,27 @@ function ResponseType (masterOptions) {
   }
 
   responseType._emitter = new EventEmitter()
+  responseType._results = []
 
   responseType.options = {
     shouldAck: !!masterOptions.shouldAck,
     shouldReply: !!masterOptions.shouldReply
   }
 
+  let wrapper = (fn) => {
+    return function () {
+      responseType._results.push(fn(...arguments))
+    }
+  }
+
   responseType.ready = function onReady (callback) {
     responseType._emitter.on('ready', callback)
 
     return responseType
   }
 
   responseType.data = function onData (callback) {
-    responseType._emitter.on('data', callback)
+    responseType._emitter.on('data', wrapper(callback))
 
     return responseType
   }
@@ -57,7 +64,14 @@ function Response (base, type, options) {
   const remit = this
 
   let response = {
-    _emitter: new EventEmitter()
+    _emitter: new EventEmitter(),
+    _results: []
+  }
+
+  let wrapper = (fn) => {
+    return function () {
+      response._results.push(fn(...arguments))
+    }
   }
 
   response.ready = function onReady (callback) {
@@ -74,7 +88,7 @@ function Response (base, type, options) {
   }
 
   response.data = function onData (callbacks) {
-    callbacks = Array.isArray(callbacks) ? callbacks : [callbacks]
+    callbacks = (Array.isArray(callbacks) ? callbacks : [callbacks]).map(wrapper)
     const finalCallback = callbacks.pop()
 
     if (!callbacks.length) {

lib/handleMessage.js

@@ -25,8 +25,28 @@ function handleMessage (type, response, message) {
   const event = parseEvent(message.properties, message.fields, messageContent)
 
   response._emitter.once('done', getCallbackHandler.apply(remit, [type, response, message, event]))
+
+  type._results = []
+  response._results = []
   type._emitter.emit('data', event, callback)
   response._emitter.emit('data', event, callback)
+
+  const results = type._results.concat(response._results)
+
+  const promises = results.reduce((list, result) => {
+    if (result && result.then && typeof result.then === 'function') {
+      list.push(result)
+    }
+
+    return list
+  }, [])
+
+  if (promises.length) {
+    Promise
+      .race(promises)
+      .then((v) => callback(null, v))
+      .catch(callback)
+  }
 }
 
 module.exports = handleMessage