dandri/zigbee2mqtt
1
0
Fork 0
mirror of https://github.com/Koenkk/zigbee2mqtt.git synced 2026-08-25 03:59:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor zigbee publish queue to zigbee.js

Browse Source
This commit is contained in:
Koen Kanters
2019-02-02 01:41:05 +01:00
parent 3963a8d26f
commit fd4a88acd0
2 changed files with 67 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/extension/devicePublish.js
+23 -45
View File
@@ -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);
}
});
lib/zigbee.js
+44 -40
View File
@@ -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);