add logic for send ack to ws ack

lib/agent/ack.js

@@ -0,0 +1,67 @@
+const storage = require('./utils/storage');
+
+    exports.processAck = function (json, cb){
+        //json.ack_id = '1234534'
+        if (!existKeyAckInJson(json)) {
+        return cb(new Error('no existe el key ack_id en el json'));
+        }
+        else{
+            exports.verifyIfExistId(json.ack_id, (err, exist) => {
+                if (err) return cb(new Error(err));
+                if (exist) return cb(new Error('El ack_id ya existe!'));
+                exports.registerAck(json, (err, register) => {
+                    if (err) return cb(new Error(err));
+                    return cb(null, register)
+                });
+            });
+        }
+    }
+
+    const existKeyAckInJson = function (json) {
+        if (json.hasOwnProperty("ack_id")) 
+            return true;
+        return false;
+    }
+    exports.updateAck = function (json, cb) {
+        storage.do('update', {
+            type: 'ack',
+            id: json.ack_id,
+            columns: ['retries'],
+            values: [json.retries + 1],
+          }, (err) => {
+            if (err) return cb(new Error(err));
+            return cb && cb(null, {
+                "ack_id": json.ack_id,
+                "type": "ack",
+                "retries": json.retries + 1
+            })
+          });
+    }
+
+     exports.registerAck = function (json, cb) {
+        storage.do(
+            'set', 
+            {type: 'ack', id: json.ack_id, data: {id: json.ack_id, type: 'ack', retries: json.retries}},
+            function(err) {
+            if (err) return cb(new Error(err));
+            return cb(null, {
+                "ack_id": json.ack_id,
+                "type": "ack"
+            })
+        })
+    }
+
+    exports.verifyIfExistId = function (ack_id, cb) {
+        storage.do(
+        'query',
+        { type: 'ack', column: 'id', data: ack_id },
+        function (err, rows) {
+            if (err) return cb(err);
+
+            if (rows && rows.length == 0)
+            return cb(null, false);
+
+            else return cb(null, true);
+        }
+        );
+    };

lib/agent/actions.js

@@ -5,8 +5,6 @@ var join    = require('path').join,
 var actions = {},
     running = {};
 
-var actions_path = __dirname + '/actions';
-
 var load_action = function(type, name) {
   var file = join(__dirname, type + 's', name);
   try {

lib/agent/plugins/control-panel/websockets/index.js

@@ -2,14 +2,14 @@ const WebSocket = require('ws');
 const HttpsProxyAgent = require('https-proxy-agent');
 const { v4: uuidv4 } = require('uuid');
 const server = require('./server');
-const urlModule = require('url');
 const { EventEmitter } = require('events');
 const keys = require('../api/keys');
 const statusTrigger = require('../../../triggers/status');
 const fileretrieval = require('../../../actions/fileretrieval');
 const triggers = require('../../../actions/triggers');
 const storage = require('../../../utils/storage');
 const errors = require('../api/errors');
+const ack = require('./../../../ack');
 
 let common = require('../../../common');
 
@@ -27,6 +27,7 @@ let emitter;
 let setAliveTimeInterval = null;
 let setIntervalWSStatus = null;
 let notifyActionInterval = null;
+let notifyAckInterval = null;
 let getStatusInterval = null;
 let idTimeoutToCancel;
 let websocketConnected = false;
@@ -37,9 +38,11 @@ let codeErrorNoConnectionWebSocket = 1006;
 
 let startupTimeout = 5000;
 let heartbeatTimeout = 120000 + 1000;
+const retriesMax = 10;
 
 exports.re_schedule = true;
 exports.responses_queue = [];
+exports.responsesAck = [];
 
 const propagateError = (message) => {
   hooks.trigger('error', new Error(message));
@@ -73,6 +76,112 @@ const retryQueuedResponses = () => {
   });
 };
 
+const retryAckResponses = () => {
+  if (exports.responsesAck.length === 0) return;
+
+  exports.responsesAck.forEach((respoAck) => {
+    exports.notifyAck(
+      respoAck.ack_id,
+      respoAck.type,
+      respoAck.retries
+    );
+  });
+}
+
+const setArrayAcks  = () => {
+  storage.do('all', { type: 'ack' }, (errs, acks) => {
+    if (!acks || typeof acks === 'undefined') return;
+
+    if (acks.length === 0 || errs) return;
+
+    if (Array.isArray(acks)) {
+      exports.responsesAck = acks.map((element) => ({
+        ack_id: element.id,
+        type: element.type,
+        retries: element.retries
+      }));
+    } else {
+      exports.responsesAck.push({
+        ack_id: acks.id,
+        type: acks.type,
+        retries: acks.retries
+      });
+    }
+  })
+}
+
+exports.sendAckToServer = (sendToWs)  => {
+  try {
+    if (!ws || !ws.readyState || ws.readyState !== 1) return;
+    const toSend = {ack_id: sendToWs.ack_id, type: sendToWs.type};
+    ws.send(JSON.stringify(toSend))
+    exports.responsesAck = exports.responsesAck.filter((x) => x.ack_id !== sendToWs.ack_id);
+    storage.do('del', { type: 'ack', id: sendToWs.ack_id });
+  } catch (error) {
+    console.log('error process sendToServer ack:', error)
+  }
+}
+
+exports.notifyAck = (ack_id, type, retries = 0) => {
+
+  if (retries >= retriesMax) {
+    storage.do('del', { type: 'ack', id: ack_id });
+    exports.responsesAck = exports.responsesAck.filter((x) => x.ack_id !== ack_id);
+    return;
+  }
+
+  let ackResponse = exports.responsesAck.filter((ack) => ack.ack_id === ack_id);
+  if (ackResponse.length == 0) {
+
+    ack.verifyIfExistId(ack_id, (err, exist) => {
+      if (err) return;
+      if(!exist) {
+        ack.registerAck({ ack_id: ack_id, type: type, retries:0}, (err, register) => {
+          if (err) return cb(new Error(err));
+          exports.responsesAck.push(register);
+          exports.sendAckToServer(register);
+      });
+      }
+      else {
+        exports.sendAckToServer({ ack_id: ack_id, type: type});      
+      }
+      })
+  }
+  else {
+    ack.verifyIfExistId(ack_id, (err, exist) => {
+      if (err) return;
+      if(!exist) {
+        ack.registerAck(ackResponse[0], (err, register) => {
+          if (err) return cb(new Error(err));
+          exports.responsesAck.push(register);
+          exports.sendAckToServer(register);
+      });
+      }
+      else {
+        ackResponse[0].retries = ackResponse[0].retries + 1;
+        ack.updateAck(ackResponse[0]);
+        exports.sendAckToServer(ackResponse[0]);      
+      }
+      })
+    }
+}
+
+const processAcks = (arr) => {
+  if (arr.forEach) {
+    arr.forEach((el) => {
+      ack.processAck(el, (err, sendToWs) => {
+        if (err) return;
+        exports.responsesAck.push({
+          ack_id: sendToWs.ack_id,
+          type: sendToWs.type,
+          retries: 0
+        });
+        return sendToWs;
+      })
+    })
+  }
+}
+
 const processCommands = (arr) => {
   if (arr.forEach) {
     arr.forEach((el) => {
@@ -118,7 +227,7 @@ const restartWebsocketCall = () => {
   if (ws) ws.terminate();
   if (exports.re_schedule) {
     if (idTimeoutToCancel) clearTimeout(idTimeoutToCancel);
-    idTimeoutToCancel = setTimeout(exports.startWebsocket, 5000);
+      idTimeoutToCancel = setTimeout(exports.startWebsocket, 5000);
   }
 }
 
@@ -161,6 +270,7 @@ const loadServer = () => {
 exports.startWebsocket = () => {
   clearAndResetIntervals();
   notifyActionInterval = setInterval(retryQueuedResponses, 5000); // <-revisar el tiempo
+  notifyAckInterval = setInterval(retryAckResponses, 5000);
   getStatusInterval = setInterval(getStatusByInterval, 5 * 60 * 1000);
   setIntervalWSStatus = setInterval(exports.heartbeat, 20000);
   const proxy = config.get('try_proxy');
@@ -187,7 +297,6 @@ exports.startWebsocket = () => {
     },
   };
 
-
   if (proxy && workingWithProxy) {
     const agent = new HttpsProxyAgent(proxy);
     options.agent = agent;
@@ -203,9 +312,12 @@ exports.startWebsocket = () => {
     ws.on('open', () => {
       pingInterval = setInterval(() => { ws.ping() }, 60000);
       exports.notify_status(status);
+
       storage.do('all', { type: 'responses' }, (errs, actions) => {
         if (!actions || typeof actions === 'undefined') return;
+
         if (actions.length === 0 || errs) return;
+
         if (Array.isArray(actions)) {
           exports.responses_queue = actions.map((element) => ({
             reply_id: `${element.action_id}`, // id de la acción
@@ -232,6 +344,9 @@ exports.startWebsocket = () => {
           });
         }
       });
+
+      setArrayAcks();
+
     });
 
     ws.on('close', (code) => {
@@ -248,17 +363,20 @@ exports.startWebsocket = () => {
       let parsedData;
       try {
         parsedData = JSON.parse(data);
+
       } catch (e) {
         return propagateError('Invalid command object');
       }
       if (Array.isArray(parsedData)) {
         const len = parsedData.length;
         if (len && len > 0) {
           processCommands(parsedData);
+          processAcks(parsedData);
         }
         return 0;
       }
       if (parsedData.status && parsedData.status === 'OK') {
+      console.log('parsedData:', parsedData)
         const value = exports.responses_queue.find((x) => x.id === parsedData.id);
         if (value) {
           if (value.type === 'response') {
@@ -284,7 +402,7 @@ exports.startWebsocket = () => {
 
 exports.notify_action = (status, id, action, opts, err, out, time, respId, retries = 0) => {
   if (!id || id === 'report' || action === 'triggers' || action === 'geofencing') return;
-  if (retries >= 10) {
+  if (retries >= retriesMax) {
     storage.do('del', { type: 'responses', id: respId });
     exports.responses_queue = exports.responses_queue.filter((x) => x.id !== respId);
     return;
@@ -399,6 +517,7 @@ exports.load = function (cb) {
 
 const clearAndResetIntervals = (aliveTimeReset = false) => {
   if(notifyActionInterval) clearInterval(notifyActionInterval);
+  if(notifyAckInterval) clearInterval(notifyAckInterval);
   if(getStatusInterval) clearInterval(getStatusInterval);
   if(setIntervalWSStatus) clearInterval(setIntervalWSStatus);
   if(pingTimeout) clearInterval(pingTimeout);

lib/agent/utils/storage.js

@@ -62,6 +62,13 @@ var types = {
     values: (data) => {
       return `'${data.id}', '${data.from}', '${data.to}','${data.attempts}','${data.notified}'`;
     }
+  },
+  ack: {
+    schema: 'id TEXT PRIMARY KEY, type TEXT, retries INTEGER',
+    keys: ['id', 'type', 'retries'],
+    values: (data) => {
+      return `'${data.id}', '${data.type}', 0`;
+    }
   }
 };
 var SQLITE_ACCESS_ERR = 'Access denied to commands database, must run agent as prey user';