Implement zigbee queue. #1027

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);
+            });
         });
     }
 }