From 733819ad4e5c17fa5c755ca6a47256339fa92c92 Mon Sep 17 00:00:00 2001
From: cheetah <git@cheetah.cat>
Date: Fri, 26 Mar 2021 19:54:04 +0000
Subject: [PATCH] added http delivery api made delivery process easier and
 optimized it

---
 index.js                   |  6 ++++++
 types/ConnectorRegistry.js |  2 +-
 types/MessageManager.js    | 38 +++++++++++++++++++++-----------------
 3 files changed, 28 insertions(+), 18 deletions(-)

diff --git a/index.js b/index.js
index 2ba94bf..fa873f8 100644
--- a/index.js
+++ b/index.js
@@ -37,8 +37,14 @@ app.post('/api/message/advanced', async (req, res) => {
     await types.MessageManager.Deliver(id)
     return res.json(id)
 })
+
 app.get('/api/message/status/:id', async (req, res) => { //TODO: make this fancy
     return res.json(types.MessageManager.messages[ req.params.id ])
 })
 
+app.get('/api/message/ack/recv/:id', async (req, res) => { //TODO: make this fancy
+    types.ConnectorRegistry.reportDelivered({ id: req.params.id }, 'http')
+    return res.json(true)
+})
+
 app.listen(3000)
\ No newline at end of file
diff --git a/types/ConnectorRegistry.js b/types/ConnectorRegistry.js
index cb9d603..c406cb0 100644
--- a/types/ConnectorRegistry.js
+++ b/types/ConnectorRegistry.js
@@ -29,7 +29,7 @@ class ConnectorRegistry {
         this.reportState(msg, uuid, 'failed')
     }
     reportDelivered(msg, uuid) {
-        this.events.emit(`msg:status:${ msg.id }:delivered`)
+        this.events.emit(`msg:status:${ msg.id }:delivered`, msg, uuid)
         this.reportState(msg, uuid, 'delivered')
     }
 
diff --git a/types/MessageManager.js b/types/MessageManager.js
index 4f4ff25..1868890 100644
--- a/types/MessageManager.js
+++ b/types/MessageManager.js
@@ -23,9 +23,9 @@ class MessageManager {
         const hwDuplexSupport = require("./DeviceRegistry").Devices[ routingParams.device ].duplex || false
         Object.assign(this.messages[ msgObj.id ]._routerData, hwDuplexSupport ? {
             duplexCapable: true,
-            recvAck: false,
-            readAck: false,
-            response: false,
+            recvAck: false, // aka "delivered"
+            readAck: false, // "read"
+            response: false, // "resp"
             deliveryLog: {},
         } : {
             duplexCapable: false,
@@ -40,6 +40,7 @@ class MessageManager {
     }
     async msgStatus(msgId, uuid, status) {
         this.messages[ msgId ]._routerData.deliveryLog[ uuid ] = status
+        if (status === 'delivered') this.messages[ msgId ]._routerData.recvAck = true
         //this.Deliver(msgId)
         console.log(msgId, uuid, 'status is', status)
     }
@@ -65,7 +66,6 @@ class MessageManager {
                     connectorConfig = config.connectors[connectorName]
             //const UUID = md5(JSON.stringify(connectorDeliveryTry))
             const chainPromise = (res) => {
-                ConnectorRegistry.events.removeAllListeners(`msg:status:${ msgId }:delivered`)
                 ConnectorRegistry.events.removeAllListeners(`msg:status:${ msgId }:failed`)
                 setTimeout(() => {
                     res([false, 'timeout'])
@@ -74,24 +74,28 @@ class MessageManager {
                     console.log(`${ msgId } failed via ${ connectorName }:${ connectorArgs.join(',') }, continuing...`)
                     res([false, 'failed'])
                 })
-                ConnectorRegistry.events.once(`msg:status:${ msgId }:delivered`, () => {
-                    console.log(`${ msgId } delivered via ${ connectorName }:${ connectorArgs.join(',') }`)
-                    res([true, 'delivered'])
-                })
                 console.log(`Trying to deliver msg#${ msg.id } with ${ connectorName }:${ connectorArgs.join(',') }`)
-                //console.log(this.messages[ msgId ].deliveryLog)
                 ConnectorRegistry.transmit(connectorName, msg, connectorArgs)
             }
             return chainPromise
         })
-        new Promise(async (res, rej) => {
-            for(let deliveryFunction of deliveryChain) {
-                let result = await new Promise(deliveryFunction)
-                if (result[0] === true) { res(result); break; }
-                // TODO: handle case, when a different verification channel is used for ACK
-            }
-            rej()
-        })
+        Promise.race([
+            new Promise(res => { // Delivery Event for this message
+                ConnectorRegistry.events.once(`msg:status:${ msgId }:delivered`, (_, uuid) => {
+                    console.log(`${ msgId } delivered via ${ uuid }`)
+                    return res()
+                })
+            }),
+            new Promise(async (res, rej) => {
+                for(let deliveryFunction of deliveryChain) {
+                    //when a different verification channel is used for ACK
+                    if (this.messages[ msgId ]._routerData.recvAck === true) break;
+                    let result = await new Promise(deliveryFunction)
+                    if (result[0] === true) { res(); break; }
+                }
+                rej()
+            })
+        ])
         .then(($) => {
             this._clearEventHandlers4MsgID(msgId)
             const dLog = this.messages[ msgId ]._routerData.deliveryLog