From 9c8323326d704dca8f213ef088e5566096efdc7b Mon Sep 17 00:00:00 2001 From: Koen Kanters Date: Mon, 16 Nov 2020 17:03:22 +0100 Subject: [PATCH] Greatly reduce amount of subscribed topics. #4948 --- lib/extension/bind.js | 14 ------------- lib/extension/bridge.js | 1 - lib/extension/configure.js | 12 ----------- lib/extension/groups.js | 21 ------------------- lib/extension/homeassistant.js | 10 ++++----- lib/extension/legacy/bridgeLegacy.js | 2 -- lib/extension/legacy/deviceGroupMembership.js | 7 ------- lib/extension/networkMap.js | 12 ----------- lib/extension/otaUpdate.js | 12 ----------- lib/extension/publish.js | 12 ----------- lib/mqtt.js | 13 ++++++++++-- test/bind.test.js | 6 ------ test/group.test.js | 6 ------ test/otaUpdate.test.js | 7 ------- 14 files changed, 16 insertions(+), 119 deletions(-) diff --git a/lib/extension/bind.js b/lib/extension/bind.js index 765e178aa..9b114fd96 100644 --- a/lib/extension/bind.js +++ b/lib/extension/bind.js @@ -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; diff --git a/lib/extension/bridge.js b/lib/extension/bridge.js index 0c1b8ad28..69d5e7348 100644 --- a/lib/extension/bridge.js +++ b/lib/extension/bridge.js @@ -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(); diff --git a/lib/extension/configure.js b/lib/extension/configure.js index f4a1b0285..62cc2c34d 100644 --- a/lib/extension/configure.js +++ b/lib/extension/configure.js @@ -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); diff --git a/lib/extension/groups.js b/lib/extension/groups.js index 2f200d5a8..2106d230d 100644 --- a/lib/extension/groups.js +++ b/lib/extension/groups.js @@ -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(); diff --git a/lib/extension/homeassistant.js b/lib/extension/homeassistant.js index 9661604ad..3b2545d8f 100644 --- a/lib/extension/homeassistant.js +++ b/lib/extension/homeassistant.js @@ -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); } diff --git a/lib/extension/legacy/bridgeLegacy.js b/lib/extension/legacy/bridgeLegacy.js index c273a7c2e..aa30b689e 100644 --- a/lib/extension/legacy/bridgeLegacy.js +++ b/lib/extension/legacy/bridgeLegacy.js @@ -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(); } diff --git a/lib/extension/legacy/deviceGroupMembership.js b/lib/extension/legacy/deviceGroupMembership.js index d282a7f68..91a81accf 100644 --- a/lib/extension/legacy/deviceGroupMembership.js +++ b/lib/extension/legacy/deviceGroupMembership.js @@ -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) { diff --git a/lib/extension/networkMap.js b/lib/extension/networkMap.js index 8ed6b2851..96ff667bb 100644 --- a/lib/extension/networkMap.js +++ b/lib/extension/networkMap.js @@ -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) { diff --git a/lib/extension/otaUpdate.js b/lib/extension/otaUpdate.js index 89eab9e5d..9ee7b4ddb 100644 --- a/lib/extension/otaUpdate.js +++ b/lib/extension/otaUpdate.js @@ -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. diff --git a/lib/extension/publish.js b/lib/extension/publish.js index fb7e1ce28..700fe3dfd 100644 --- a/lib/extension/publish.js +++ b/lib/extension/publish.js @@ -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) { diff --git a/lib/mqtt.js b/lib/mqtt.js index 17d54811a..0efe22749 100644 --- a/lib/mqtt.js +++ b/lib/mqtt.js @@ -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!`); diff --git a/test/bind.test.js b/test/bind.test.js index b776a673a..99b9a230a 100644 --- a/test/bind.test.js +++ b/test/bind.test.js @@ -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); diff --git a/test/group.test.js b/test/group.test.js index 13efce2bb..ed5cefa29 100644 --- a/test/group.test.js +++ b/test/group.test.js @@ -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 () => { diff --git a/test/otaUpdate.test.js b/test/otaUpdate.test.js index 41075b8f3..b270a6bb5 100644 --- a/test/otaUpdate.test.js +++ b/test/otaUpdate.test.js @@ -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];