From 4a664cf230a8804f9f2fbcce00d311e7105d4e07 Mon Sep 17 00:00:00 2001
From: Koen Kanters <koenkanters94@gmail.com>
Date: Fri, 22 Feb 2019 18:58:50 +0100
Subject: [PATCH] Implement zigbee queue. #1027

---
 lib/util/zigbeeQueue.js | 84 +++++++++++++++++++++++++++++++++++++++++
 lib/zigbee.js           | 32 +++++++---------
 2 files changed, 98 insertions(+), 18 deletions(-)
 create mode 100644 lib/util/zigbeeQueue.js

diff --git a/lib/util/zigbeeQueue.js b/lib/util/zigbeeQueue.js
new file mode 100644
index 00000000..dc63b5f3
--- /dev/null
+++ b/lib/util/zigbeeQueue.js
@@ -0,0 +1,84 @@
+const maxSimultaneouslyRunning = 5;
+const delay = 250;
+const error17Retries = 3;
+
+class ZigbeeQueue {
+    constructor() {
+        this.queue = [];
+        this.active = [];
+        this.timer = null;
+        this.running = false;
+    }
+
+    start() {
+        this.running = true;
+        this.resetTimer();
+    }
+
+    push(entityID, func) {
+        this.queue.push({entityID, func, attempts: 0});
+    }
+
+    stopTimer() {
+        if (this.timer) {
+            clearInterval(this.timer);
+            this.timer = null;
+        }
+    }
+
+    resetTimer() {
+        this.stopTimer();
+
+        this.timer = setInterval(() => {
+            this.executeNext();
+        }, delay);
+    }
+
+    stop() {
+        this.stopTimer();
+        this.running = false;
+    }
+
+    handleJobComplete(job, error) {
+        if (error && error.message === 'rsp error: 17' && job.attempts < error17Retries) {
+            // Error 17 means that the buffer of the ZNP was full,
+            // retry this for a maximum of 3 times.
+            job.attempts++;
+            this.active.splice(this.active.indexOf(job), 1);
+        } else {
+            this.active.splice(this.active.indexOf(job), 1);
+            this.queue.splice(this.queue.indexOf(job), 1);
+        }
+    }
+
+    executeNext() {
+        if (!this.running) {
+            return;
+        }
+
+        const next = this.getNext();
+
+        if (next) {
+            this.active.push(next);
+            next.func((error) => this.handleJobComplete(next, error));
+        }
+    }
+
+    getNext() {
+        if (this.active.length > (maxSimultaneouslyRunning - 1)) {
+            return null;
+        }
+
+        for (let i = 0; i < this.queue.length; i++) {
+            const job = this.queue[i];
+            const activeDeviceJob = this.active.find((j) => j.entityID === job.entityID);
+            if (!activeDeviceJob) {
+                return job;
+            }
+        }
+
+        return null;
+    }
+}
+
+module.exports = ZigbeeQueue;
diff --git a/lib/zigbee.js b/lib/zigbee.js
index bf5ba3f5..d74aac2a 100644
--- a/lib/zigbee.js
+++ b/lib/zigbee.js
@@ -3,8 +3,8 @@ const logger = require('./util/logger');
 const settings = require('./util/settings');
 const data = require('./util/data');
 const utils = require('./util/utils');
+const ZigbeeQueue = require('./util/zigbeeQueue');
 const cieApp = require('./zapp/cie');
-const Queue = require('queue');
 const zclId = require('zcl-id');
 
 const advancedSettings = settings.get().advanced;
@@ -32,8 +32,6 @@ const defaultCfg = {
     disDefaultRsp: 0,
 };
 
-const delay = 170;
-
 logger.debug(`Using zigbee-shepherd with settings: '${JSON.stringify(shepherdSettings)}'`);
 
 class Zigbee {
@@ -43,8 +41,7 @@ class Zigbee {
         this.onError = this.onError.bind(this);
         this.messageHandler = null;
 
-        this.queue = new Queue();
-        this.queue.concurrency = 1;
+        this.queue = new ZigbeeQueue();
     }
 
     start(messageHandler, callback) {
@@ -120,7 +117,6 @@ class Zigbee {
 
         // Wait some time before we start the queue, many calls skip this queue which hangs the stick
         setTimeout(() => {
-            this.queue.autostart = true;
             this.queue.start();
         }, 2000);
 
@@ -240,7 +236,7 @@ class Zigbee {
             return;
         }
 
-        this.queue.push((queueCallback) => {
+        this.queue.push(entityID, (queueCallback) => {
             logger.info(
                 `Zigbee publish to ${entityType} '${entityID}', ${cid} - ${cmd} - ` +
                 `${JSON.stringify(zclData)} - ${JSON.stringify(cfg)} - ${ep}`
@@ -257,6 +253,8 @@ class Zigbee {
                 if (callback) {
                     callback(error, rsp);
                 }
+
+                queueCallback(error);
             };
 
             if (cmdType === 'functional' && entity.functional) {
@@ -266,8 +264,6 @@ class Zigbee {
             } else {
                 logger.error(`Unknown zigbee publish cmdType ${cmdType}`);
             }
-
-            setTimeout(() => queueCallback(), delay);
         });
     }
 
@@ -275,7 +271,7 @@ class Zigbee {
         const device = this.shepherd._findDevByAddr(ieeeAddr);
 
         if (device) {
-            this.queue.push((queueCallback) => {
+            this.queue.push(ieeeAddr, (queueCallback) => {
                 logger.debug(`Ping ${ieeeAddr}`);
                 this.shepherd.controller.checkOnline(device, (error) => {
                     if (error) {
@@ -287,9 +283,9 @@ class Zigbee {
                     if (cb) {
                         cb(error);
                     }
-                });
 
-                setTimeout(() => queueCallback(), delay);
+                    queueCallback(error);
+                });
             });
         }
     }
@@ -297,7 +293,7 @@ class Zigbee {
     bind(ep, cluster, target=this.getCoordinator()) {
         const log = `for ${ep.device.ieeeAddr} - ${cluster}`;
 
-        this.queue.push((queueCallback) => {
+        this.queue.push(ep.device.ieeeAddr, (queueCallback) => {
             logger.debug(`Setup binding ${log}`);
             ep.bind(cluster, target, (error) => {
                 if (error) {
@@ -305,9 +301,9 @@ class Zigbee {
                 } else {
                     logger.debug(`Successfully setup binding ${log}`);
                 }
-            });
 
-            setTimeout(() => queueCallback(), delay);
+                queueCallback(error);
+            });
         });
     }
 
@@ -317,7 +313,7 @@ class Zigbee {
         const cfg = {direction: 0, attrId, dataType, minRepIntval: min, maxRepIntval: max, repChange: change};
         const log = `for ${ep.device.ieeeAddr} - ${cluster} - ${attribute}`;
 
-        this.queue.push((queueCallback) => {
+        this.queue.push(ep.device.ieeeAddr, (queueCallback) => {
             logger.debug(`Setup reporting ${log}`);
             ep.foundation(cluster, 'configReport', [cfg], defaultCfg, (error) => {
                 if (error) {
@@ -325,9 +321,9 @@ class Zigbee {
                 } else {
                     logger.debug(`Successfully setup reporting ${log}`);
                 }
-            });
 
-            setTimeout(() => queueCallback(), delay);
+                queueCallback(error);
+            });
         });
     }
 }