From fd4a88acd06ae02e7444a5201d3160d1633bf6b7 Mon Sep 17 00:00:00 2001 From: Koen Kanters Date: Sat, 2 Feb 2019 01:41:05 +0100 Subject: [PATCH] Refactor zigbee publish queue to zigbee.js --- lib/extension/devicePublish.js | 68 ++++++++++----------------- lib/zigbee.js | 84 ++++++++++++++++++---------------- 2 files changed, 67 insertions(+), 85 deletions(-) diff --git a/lib/extension/devicePublish.js b/lib/extension/devicePublish.js index 24f95e3af..7ba17ebc2 100644 --- a/lib/extension/devicePublish.js +++ b/lib/extension/devicePublish.js @@ -1,7 +1,6 @@ const settings = require('../util/settings'); const zigbeeShepherdConverters = require('zigbee-shepherd-converters'); -const Queue = require('queue'); const logger = require('../util/logger'); const utils = require('../util/utils'); @@ -23,15 +22,6 @@ class DevicePublish { this.mqtt = mqtt; this.state = state; this.publishDeviceState = publishDeviceState; - - /** - * Setup command queue. - * The command queue ensures that only 1 command is executed at a time. - * When executing multiple commands at the same time, some commands may fail. - */ - this.queue = new Queue(); - this.queue.concurrency = 1; - this.queue.autostart = true; } onMQTTConnected() { @@ -43,10 +33,6 @@ class DevicePublish { } } - stop() { - this.queue.stop(); - } - parseTopic(topic) { if (!topic.match(topicRegex)) { return null; @@ -149,31 +135,26 @@ class DevicePublish { return; } - // Add job to queue - this.queue.push((queueCallback) => { - this.zigbee.publish( - entity.ID, - entity.type, - converted.cid, - converted.cmd, - converted.cmdType, - converted.zclData, - converted.cfg, - endpoint, - (error, rsp) => { - // Devices do not report when they go off, this ensures state (on/off) is always in sync. - if (entity.type === 'device' && topic.type === 'set' && - !error && (key.startsWith('state') || key.startsWith('brightness'))) { - const msg = {}; - const _key = topic.postfix ? `state_${topic.postfix}` : 'state'; - msg[_key] = key.startsWith('brightness') ? 'ON' : json['state']; - this.publishDeviceState(device, msg, true); - } + this.zigbee.publish( + entity.ID, + entity.type, + converted.cid, + converted.cmd, + converted.cmdType, + converted.zclData, + converted.cfg, + endpoint, + (error, rsp) => { + // Devices do not report when they go off, this ensures state (on/off) is always in sync. + if (entity.type === 'device' && topic.type === 'set' && + !error && (key.startsWith('state') || key.startsWith('brightness'))) { + const msg = {}; + const _key = topic.postfix ? `state_${topic.postfix}` : 'state'; + msg[_key] = key.startsWith('brightness') ? 'ON' : json['state']; + this.publishDeviceState(device, msg, true); } - ); - - setTimeout(() => queueCallback(), 170); - }); + } + ); // It's possible for devices to get out of sync when writing an attribute that's not reportable. // So here we re-read the value after a specified timeout, this timeout could for example be the @@ -182,13 +163,10 @@ class DevicePublish { && converted.hasOwnProperty('readAfterWriteTime') && converted.readAfterWriteTime !== 0) { const getConverted = converter.convert(key, json[key], json, 'get'); setTimeout(() => { - // Add job to queue - this.queue.push((queueCallback) => { - this.zigbee.publish( - entity.ID, entity.type, getConverted.cid, getConverted.cmd, getConverted.cmdType, - getConverted.zclData, getConverted.cfg, endpoint, () => queueCallback() - ); - }); + this.zigbee.publish( + entity.ID, entity.type, getConverted.cid, getConverted.cmd, getConverted.cmdType, + getConverted.zclData, getConverted.cfg, endpoint, () => {} + ); }, converted.readAfterWriteTime); } }); diff --git a/lib/zigbee.js b/lib/zigbee.js index 5c13e28d4..b126e45d3 100644 --- a/lib/zigbee.js +++ b/lib/zigbee.js @@ -199,46 +199,6 @@ class Zigbee { return this.shepherd.getGroup(ID); } - publish(entityID, entityType, cid, cmd, cmdType, zclData, cfg=defaultCfg, ep, callback) { - let entity = null; - if (entityType === 'device') { - entity = this.getEndpoint(entityID, ep); - } else if (entityType === 'group') { - entity = this.getGroup(entityID); - } - - if (!entity) { - logger.error( - `Zigbee cannot publish message to ${entityType} because '${entityID}' not known by zigbee-shepherd` - ); - return; - } - - logger.info( - `Zigbee publish to ${entityType} '${entityID}', ${cid} - ${cmd} - ` + - `${JSON.stringify(zclData)} - ${JSON.stringify(cfg)} - ${ep}` - ); - - const callback_ = (error, rsp) => { - if (error) { - logger.error( - `Zigbee publish to ${entityType} '${entityID}', ${cid} - ${cmd} - ${JSON.stringify(zclData)} ` + - `- ${JSON.stringify(cfg)} - ${ep} ` + - `failed with error ${error}`); - } - - callback(error, rsp); - }; - - if (cmdType === 'functional' && entity.functional) { - entity.functional(cid, cmd, zclData, cfg, callback_); - } else if (cmdType === 'foundation' && entity.foundation) { - entity.foundation(cid, cmd, zclData, cfg, callback_); - } else { - logger.error(`Unknown zigbee publish cmdType ${cmdType}`); - } - } - networkScan(callback) { logger.info('Starting network scan...'); this.shepherd.lqiScan().then((result) => { @@ -261,6 +221,50 @@ class Zigbee { return endpoint; } + publish(entityID, entityType, cid, cmd, cmdType, zclData, cfg=defaultCfg, ep, callback) { + let entity = null; + if (entityType === 'device') { + entity = this.getEndpoint(entityID, ep); + } else if (entityType === 'group') { + entity = this.getGroup(entityID); + } + + if (!entity) { + logger.error( + `Cannot publish message to ${entityType} because '${entityID}' is not known by zigbee-shepherd` + ); + return; + } + + this.queue.push((queueCallback) => { + logger.info( + `Zigbee publish to ${entityType} '${entityID}', ${cid} - ${cmd} - ` + + `${JSON.stringify(zclData)} - ${JSON.stringify(cfg)} - ${ep}` + ); + + const callback_ = (error, rsp) => { + if (error) { + logger.error( + `Zigbee publish to ${entityType} '${entityID}', ${cid} - ${cmd} - ${JSON.stringify(zclData)} ` + + `- ${JSON.stringify(cfg)} - ${ep} ` + + `failed with error ${error}`); + } + + callback(error, rsp); + }; + + if (cmdType === 'functional' && entity.functional) { + entity.functional(cid, cmd, zclData, cfg, callback_); + } else if (cmdType === 'foundation' && entity.foundation) { + entity.foundation(cid, cmd, zclData, cfg, callback_); + } else { + logger.error(`Unknown zigbee publish cmdType ${cmdType}`); + } + + setTimeout(() => queueCallback(), delay); + }); + } + ping(ieeeAddr, cb) { const device = this.shepherd._findDevByAddr(ieeeAddr);