Event acking

src/Connection/index.js

@@ -220,6 +220,22 @@ class Connection extends Emittery {
       return
     }
 
+    /**
+     * Ack packet
+     */
+    if (msp.isAckPacket(packet)) {
+      this._processAck(packet)
+      return
+    }
+
+    /**
+     * Ack error packet
+     */
+    if (msp.isAckErrorPacket(packet)) {
+      this._processAckError(packet)
+      return
+    }
+
     /**
      * Ping from client
      */
@@ -247,12 +263,73 @@ class Connection extends Emittery {
       return
     }
 
-    if (!this.hasSubscription(packet.d.topic)) {
+    const { topic, id } = packet.d
+
+    if (!this.hasSubscription(topic)) {
       this._notifyPacketDropped('_processEvent', 'dropping event since there are no subscription %j', packet)
       return
     }
 
-    this.getSubscription(packet.d.topic).serverMessage(packet.d)
+    const result = this.getSubscription(topic).serverMessage(packet.d)
+
+    if (typeof (id) !== 'undefined') {
+      result
+        .then((responses) => {
+          const data = responses.find((response) => typeof (response) !== 'undefined')
+          this.sendAckPacket(topic, id, data)
+        })
+        .catch((error) => {
+          this.sendAckErrorPacket(topic, id, error.message)
+        })
+    }
+  }
+
+  /**
+   * Processes the ack by ensuring the packet is valid and there
+   * is a subscription for the given topic.
+   *
+   * @method _processAck
+   *
+   * @param  {Object}      packet
+   *
+   * @return {void}
+   */
+  _processAck (packet) {
+    if (!msp.isValidAckPacket(packet)) {
+      this._notifyPacketDropped('_processAck', 'dropping ack since packet is invalid %j', packet)
+      return
+    }
+
+    if (!this.hasSubscription(packet.d.topic)) {
+      this._notifyPacketDropped('_processAck', 'dropping ack since there are no subscription %j', packet)
+      return
+    }
+
+    this.getSubscription(packet.d.topic).serverAck(packet.d)
+  }
+
+  /**
+   * Processes the ack error by ensuring the packet is valid and there
+   * is a subscription for the given topic.
+   *
+   * @method _processAckError
+   *
+   * @param  {Object}      packet
+   *
+   * @return {void}
+   */
+  _processAckError (packet) {
+    if (!msp.isValidAckErrorPacket(packet)) {
+      this._notifyPacketDropped('_processAckError', 'dropping ack error since packet is invalid %j', packet)
+      return
+    }
+
+    if (!this.hasSubscription(packet.d.topic)) {
+      this._notifyPacketDropped('_processAckError', 'dropping ack error since there are no subscription %j', packet)
+      return
+    }
+
+    this.getSubscription(packet.d.topic).serverAckError(packet.d)
   }
 
   /**
@@ -605,6 +682,38 @@ class Connection extends Emittery {
     this.sendPacket(msp.leavePacket(topic))
   }
 
+  /**
+   * Sends the ack packet, when the client requested the
+   * response for given event.
+   *
+   * @method sendAckPacket
+   *
+   * @param  {String}  topic
+   * @param  {Number}  id
+   * @param  {Mixed}   data
+   *
+   * @return {void}
+   */
+  sendAckPacket (topic, id, data) {
+    this.sendPacket(msp.ackPacket(topic, id, data))
+  }
+
+  /**
+   * Sends the ack error packet, when the client requested the
+   * response for given event.
+   *
+   * @method sendAckErrorPacket
+   *
+   * @param  {String}  topic
+   * @param  {Number}  id
+   * @param  {String}  message
+   *
+   * @return {void}
+   */
+  sendAckErrorPacket (topic, id, message) {
+    this.sendPacket(msp.ackErrorPacket(topic, id, message))
+  }
+
   /**
    * Makes the event packet from the topic and the
    * body
@@ -614,10 +723,11 @@ class Connection extends Emittery {
    * @param  {String}        topic
    * @param  {String}        event
    * @param  {Mixed}         data
+   * @param  {Number}        [id]
    *
    * @return {Object}
    */
-  makeEventPacket (topic, event, data) {
+  makeEventPacket (topic, event, data, id) {
     if (!topic) {
       throw new Error('Cannot send event without a topic')
     }
@@ -626,7 +736,7 @@ class Connection extends Emittery {
       throw new Error(`Topic ${topic} doesn't have any active subscriptions`)
     }
 
-    return msp.eventPacket(topic, event, data)
+    return msp.eventPacket(topic, event, data, id)
   }
 
   /**
@@ -638,11 +748,12 @@ class Connection extends Emittery {
    * @param  {String}    event
    * @param  {Mixed}     data
    * @param  {Function}  [ack]
+   * @param  {Number}    [id]
    *
    * @return {void}
    */
-  sendEvent (topic, event, data, ack) {
-    this.sendPacket(this.makeEventPacket(topic, event, data), {}, ack)
+  sendEvent (topic, event, data, ack, id) {
+    this.sendPacket(this.makeEventPacket(topic, event, data, id), {}, ack)
   }
 
   /**

src/Socket/index.js

@@ -26,6 +26,9 @@ const ClusterHop = require('../ClusterHop')
  */
 class Socket {
   constructor (topic, connection) {
+    this._acks = new Map()
+    this._nextAckId = 0
+
     this.channel = null
 
     /**
@@ -109,11 +112,18 @@ class Socket {
    * @param  {String}   event
    * @param  {Object}   data
    * @param  {Function} [ack]
+   * @param  {Function} [responseAck]
    *
    * @return {void}
    */
-  emit (event, data, ack) {
-    this.connection.sendEvent(this.topic, event, data, ack)
+  emit (event, data, ack, responseAck) {
+    let id
+    if (typeof (responseAck) === 'function') {
+      id = this._nextAckId++
+      this._acks.set(id, responseAck)
+    }
+
+    this.connection.sendEvent(this.topic, event, data, ack, id)
   }
 
   /**
@@ -222,10 +232,50 @@ class Socket {
    * @param  {String}      options.event
    * @param  {Mixed}       options.data
    *
-   * @return {void}
+   * @return {Promise}
    */
   serverMessage ({ event, data }) {
-    this.emitter.emit(event, data)
+    return this.emitter.emit(event, data)
+  }
+
+  /**
+   * A new ack received
+   *
+   * @method serverAck
+   *
+   * @param  {Number}      options.id
+   * @param  {Mixed}       options.data
+   *
+   * @return {void}
+   */
+  serverAck ({ id, data }) {
+    if (this._acks.has(id)) {
+      const ack = this._acks.get(id)
+      ack(null, data)
+      this._acks.delete(id)
+    } else {
+      debug('bad ack %s for %s topic', id, this.topic)
+    }
+  }
+
+  /**
+   * A new ack error received
+   *
+   * @method serverAckError
+   *
+   * @param  {Number}      options.id
+   * @param  {String}      options.message
+   *
+   * @return {void}
+   */
+  serverAckError ({ id, message }) {
+    if (this._acks.has(id)) {
+      const ack = this._acks.get(id)
+      ack(new Error(message))
+      this._acks.delete(id)
+    } else {
+      debug('bad ack %s for %s topic', id, this.topic)
+    }
   }
 
   /**
@@ -241,9 +291,11 @@ class Socket {
       .emit('close', this)
       .then(() => {
         this.emitter.clearListeners()
+        this._acks.clear()
       })
       .catch(() => {
         this.emitter.clearListeners()
+        this._acks.clear()
       })
   }