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

Greatly reduce amount of subscribed topics. #4948

Browse Source
This commit is contained in:
Koen Kanters
2020-11-16 17:03:22 +01:00
parent da3b3b565f
commit 9c8323326d
14 changed files with 16 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/extension/bind.js
-14
View File
@@ -17,20 +17,6 @@ class Bind extends Extension {
this.legacyApi = settings.get().advanced.legacy_api;
}
onMQTTConnected() {
/* istanbul ignore else */
if (this.legacyApi) {
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/bind/#`);
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/unbind/#`);
}
/* istanbul ignore else */
if (settings.get().experimental.new_api) {
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/request/device/bind`);
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/request/device/unbind`);
}
}
parseMQTTMessage(topic, message) {
let type = null;
let sourceKey = null;
lib/extension/bridge.js
-1
View File
@@ -42,7 +42,6 @@ class Bridge extends Extension {
this.eventBus.on(`groupMembersChanged`, () => this.publishGroups());
this.eventBus.on(`devicesChanged`, () => this.publishDevices());
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/request/#`);
await this.publishInfo();
await this.publishDevices();
await this.publishGroups();
lib/extension/configure.js
-12
View File
@@ -53,18 +53,6 @@ class Configure extends Extension {
return true;
}
onMQTTConnected() {
/* istanbul ignore else */
if (settings.get().advanced.legacy_api) {
this.mqtt.subscribe(this.legacyTopic);
}
/* istanbul ignore else */
if (settings.get().experimental.new_api) {
this.mqtt.subscribe(this.topic);
}
}
async onMQTTMessage(topic, message) {
if (topic === this.legacyTopic) {
const resolvedEntity = this.zigbee.resolveEntity(message);
lib/extension/groups.js
-21
View File
@@ -19,27 +19,6 @@ class Groups extends Extension {
this.lastOptimisticState = {};
}
onMQTTConnected() {
/* istanbul ignore else */
if (this.legacyApi) {
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/group/remove_all`);
for (let step = 1; step < 20; step++) {
const topic = `${settings.get().mqtt.base_topic}/bridge/group/${'+/'.repeat(step)}`;
this.mqtt.subscribe(`${topic}remove`);
this.mqtt.subscribe(`${topic}add`);
this.mqtt.subscribe(`${topic}remove_all`);
}
}
/* istanbul ignore else */
if (settings.get().experimental.new_api) {
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/request/group/members/add`);
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/request/group/members/remove`);
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/request/group/members/remove_all`);
}
}
async onZigbeeStarted() {
this.eventBus.on('stateChange', this.onStateChange);
await this.syncGroupsWithSettings();
lib/extension/homeassistant.js
+5 -5
View File
@@ -459,7 +459,7 @@ class HomeAssistant extends Extension {
delete this.discovered[resolvedEntity.device.ieeeAddr];
for (const config of this.getConfigs(resolvedEntity)) {
const topic = this.getDiscoveryTopic(config, resolvedEntity.device);
this.mqtt.publish(topic, null, {retain: true, qos: 0}, this.discoveryTopic);
this.mqtt.publish(topic, null, {retain: true, qos: 0}, this.discoveryTopic, false, false);
}
}
@@ -561,7 +561,7 @@ class HomeAssistant extends Extension {
if (homeAssisantRename) {
for (const config of this.getConfigs(resolvedEntity)) {
const topic = this.getDiscoveryTopic(config, device);
this.mqtt.publish(topic, null, {retain: true, qos: 0}, this.discoveryTopic);
this.mqtt.publish(topic, null, {retain: true, qos: 0}, this.discoveryTopic, false, false);
}
}
@@ -803,7 +803,7 @@ class HomeAssistant extends Extension {
}
const topic = this.getDiscoveryTopic(config, device);
this.mqtt.publish(topic, stringify(payload), {retain: true, qos: 0}, this.discoveryTopic);
this.mqtt.publish(topic, stringify(payload), {retain: true, qos: 0}, this.discoveryTopic, false, false);
});
this.discovered[device.ieeeAddr] = true;
@@ -855,7 +855,7 @@ class HomeAssistant extends Extension {
if (clear) {
logger.debug(`Clearing Home Assistant config '${topic}'`);
topic = topic.substring(this.discoveryTopic.length + 1);
this.mqtt.publish(topic, null, {retain: true, qos: 0}, this.discoveryTopic);
this.mqtt.publish(topic, null, {retain: true, qos: 0}, this.discoveryTopic, false, false);
}
} else if ((topic === this.statusTopic || topic === defaultStatusTopic) && message.toLowerCase() === 'online') {
const timer = setTimeout(async () => {
@@ -913,7 +913,7 @@ class HomeAssistant extends Extension {
device: this.getDevicePayload(entity),
};
await this.mqtt.publish(topic, stringify(payload), {retain: true, qos: 0}, this.discoveryTopic);
await this.mqtt.publish(topic, stringify(payload), {retain: true, qos: 0}, this.discoveryTopic, false, false);
this.discoveredTriggers[device.ieeeAddr].add(discoveredKey);
}
lib/extension/legacy/bridgeLegacy.js
-2
View File
@@ -351,8 +351,6 @@ class BridgeLegacy extends Extension {
}
async onMQTTConnected() {
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/config/+`);
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/config/+/+`);
await this.publish();
}
lib/extension/legacy/deviceGroupMembership.js
-7
View File
@@ -6,13 +6,6 @@ const Extension = require('../extension');
const topicRegex = new RegExp(`^${settings.get().mqtt.base_topic}/bridge/device/(.+)/get_group_membership$`);
class DeviceGroupMembership extends Extension {
onMQTTConnected() {
for (let step = 1; step < 20; step++) {
const topic = `${settings.get().mqtt.base_topic}/bridge/device/${'+/'.repeat(step)}get_group_membership`;
this.mqtt.subscribe(topic);
}
}
async onMQTTMessage(topic, message) {
const match = topic.match(topicRegex);
if (!match) {
lib/extension/networkMap.js
-12
View File
@@ -29,18 +29,6 @@ class NetworkMap extends Extension {
};
}
onMQTTConnected() {
/* istanbul ignore else */
if (this.legacyApi) {
this.mqtt.subscribe(this.legacyTopic);
this.mqtt.subscribe(this.legacyTopicRoutes);
}
if (settings.get().experimental.new_api) {
this.mqtt.subscribe(this.topic);
}
}
async onMQTTMessage(topic, message) {
/* istanbul ignore else */
if (this.legacyApi) {
lib/extension/otaUpdate.js
-12
View File
@@ -18,18 +18,6 @@ class OTAUpdate extends Extension {
}
onMQTTConnected() {
/* istanbul ignore else */
if (this.legacyApi) {
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/ota_update/check`);
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/ota_update/update`);
}
/* istanbul ignore else */
if (settings.get().experimental.new_api) {
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/request/device/ota_update/check`);
this.mqtt.subscribe(`${settings.get().mqtt.base_topic}/bridge/request/device/ota_update/update`);
}
for (const device of this.zigbee.getClients()) {
// In case Zigbee2MQTT is restared during an update, progress and remaining values are still in state.
// remove them.
lib/extension/publish.js
-12
View File
@@ -27,18 +27,6 @@ const groupConverters = [
];
class EntityPublish extends Extension {
onMQTTConnected() {
// Subscribe to topics.
const baseTopic = settings.get().mqtt.base_topic;
for (let step = 1; step < 20; step++) {
const topic = `${baseTopic}/${'+/'.repeat(step)}`;
this.mqtt.subscribe(`${topic}set`);
this.mqtt.subscribe(`${topic}set/+`);
this.mqtt.subscribe(`${topic}get`);
this.mqtt.subscribe(`${topic}get/+`);
}
}
parseTopic(topic) {
const match = topic.match(topicRegex);
if (!match) {
lib/mqtt.js
+11 -2
View File
@@ -8,6 +8,7 @@ class MQTT extends events.EventEmitter {
constructor() {
super();
this.onMessage = this.onMessage.bind(this);
this.publishedTopics = new Set();
}
async connect() {
@@ -75,6 +76,7 @@ class MQTT extends events.EventEmitter {
const self = this;
this.client.on('connect', () => {
logger.info('Connected to MQTT server');
self.subscribe(`${settings.get().mqtt.base_topic}/#`);
self.publish('bridge/state', 'online', {retain: true, qos: 0});
resolve();
});
@@ -96,17 +98,24 @@ class MQTT extends events.EventEmitter {
}
onMessage(topic, message) {
this.emit('message', {topic, message: message + ''});
// Since we subscribe to zigbee2mqtt/# we also receive the message we send ourselves, skip these.
if (!this.publishedTopics.has(topic)) {
this.emit('message', {topic, message: message + ''});
}
}
isConnected() {
return this.client && !this.client.reconnecting;
}
async publish(topic, payload, options, base=settings.get().mqtt.base_topic, skipLog=false) {
async publish(topic, payload, options, base=settings.get().mqtt.base_topic, skipLog=false, skipReceive=true) {
topic = `${base}/${topic}`;
options = {qos: 0, retain: false, ...options};
if (skipReceive) {
this.publishedTopics.add(topic);
}
if (!this.isConnected()) {
if (!skipLog) {
logger.error(`Not connected to MQTT server!`);
test/bind.test.js
-6
View File
@@ -33,12 +33,6 @@ describe('Bind', () => {
MQTT.publish.mockClear();
});
it('Should subscribe to topics', async () => {
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/request/device/bind');
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/request/device/unbind');
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/bind/#');
});
it('Should bind', async () => {
const device = zigbeeHerdsman.devices.remote;
const target = zigbeeHerdsman.devices.bulb_color.getEndpoint(1);
test/group.test.js
-6
View File
@@ -127,12 +127,6 @@ describe('Groups', () => {
expect(group.members).toStrictEqual([endpoint]);
expect(settings.getGroup('group/with/slashes').devices).toStrictEqual([`${device.ieeeAddr}/1`]);
expect(MQTT.publish).toHaveBeenCalledWith("zigbee2mqtt/bridge/log", stringify({"type":"device_group_add","message":{"friendly_name":"bulb_color","group":"group/with/slashes"}}), {"retain": false, qos: 0}, expect.any(Function));
// Test if subscribed to topics with slashes
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/group/+/remove');
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/group/+/+/remove');
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/group/+/+/+/+/+/remove');
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/group/+/+/+/+/+/add');
});
it('Legacy api: Add to group via MQTT with postfix', async () => {
test/otaUpdate.test.js
-7
View File
@@ -27,13 +27,6 @@ describe('OTA update', () => {
MQTT.publish.mockClear();
});
it('Should subscribe to topics', async () => {
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/ota_update/check');
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/ota_update/update');
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/request/device/ota_update/check');
expect(MQTT.subscribe).toHaveBeenCalledWith('zigbee2mqtt/bridge/request/device/ota_update/update');
});
it('Should OTA update a device', async () => {
const device = zigbeeHerdsman.devices.bulb;
const endpoint = device.endpoints[0];