feat: send ack message when notify is actually completed

deepstreamIO · Jun 28, 2019 · 46b027c · 46b027c
1 parent 01105c5
commit 46b027c
Showing 1 changed file with 20 additions and 6 deletions.
diff --git a/src/handlers/record/record-handler.ts b/src/handlers/record/record-handler.ts
@@ -150,16 +150,17 @@ export default class RecordHandler implements Handler<RecordMessage> {
     }
 
     if (message.action === RA.NOTIFY) {
-      this.services.clusterNode.send(message)
       this.recordUpdatedWithoutDeepstream(message, socketWrapper)
+      this.services.clusterNode.send(message)
       return
     }
 
     this.services.logger.error(PARSER_ACTIONS[PARSER_ACTIONS.UNKNOWN_ACTION], RA[action], this.metaData)
   }
 
   private recordUpdatedWithoutDeepstream (message: RecordMessage, socketWrapper: SocketWrapper | null = null) {
-    message.names!.forEach((recordName) => {
+    let completed = 0
+    message.names!.forEach((recordName, index, names) => {
       if (this.subscriptionRegistry.hasLocalSubscribers(recordName)) {
         this.recordRequest(recordName, null, (name: string, version: number, data: JSONObject) => {
           if (version === -1) {
@@ -177,12 +178,25 @@ export default class RecordHandler implements Handler<RecordMessage> {
               parsedData: data
             }, true, null)
           }
-        }, onRequestError, message)
+
+          completed++
+          if (completed === names.length && socketWrapper) {
+            socketWrapper.sendAckMessage(message)
+          }
+        }, (event: RA, errorMessage: string, name: string, socket: SocketWrapper, msg: Message) => {
+          completed++
+          if (completed === names.length && socketWrapper) {
+            socketWrapper.sendAckMessage(message)
+          }
+          onRequestError(event, errorMessage, recordName, socket, msg)
+        }, message)
+      } else {
+        completed++
+        if (completed === names.length && socketWrapper) {
+          socketWrapper.sendAckMessage(message)
+        }
       }
     })
-    if (socketWrapper) {
-      socketWrapper.sendAckMessage(message)
-    }
   }
 
 /**