diff --git a/lib/mqtt.ts b/lib/mqtt.ts index 9a3fefd81..97282d28b 100644 --- a/lib/mqtt.ts +++ b/lib/mqtt.ts @@ -10,6 +10,8 @@ export default class MQTT { private connectionTimer: NodeJS.Timeout; private client: mqtt.MqttClient; private eventBus: EventBus; + private initialConnect = true; + private republishRetainedTimer: NodeJS.Timer; private retainedMessages: {[s: string]: {payload: string, options: MQTTOptions, skipLog: boolean, skipReceive: boolean, topic: string, base: string}} = {}; @@ -97,13 +99,19 @@ export default class MQTT { logger.info('Connected to MQTT server'); - // Republish retained messages in case MQTT broker does not persist them. - // https://github.com/Koenkk/zigbee2mqtt/issues/9629 - Object.values(this.retainedMessages).forEach((e) => - this.publish(e.topic, e.payload, e.options, e.base, e.skipLog, e.skipReceive)); + if (this.initialConnect) { + await this.publishStateOnline(); + } else { + this.republishRetainedTimer = setTimeout(() => { + // Republish retained messages in case MQTT broker does not persist them. + // https://github.com/Koenkk/zigbee2mqtt/issues/9629 + Object.values(this.retainedMessages).forEach((e) => + this.publish(e.topic, e.payload, e.options, e.base, e.skipLog, e.skipReceive)); + }, 2000); + } + this.initialConnect = false; this.subscribe(`${settings.get().mqtt.base_topic}/#`); - await this.publishStateOnline(); } async publishStateOnline(): Promise { @@ -129,6 +137,11 @@ export default class MQTT { logger.debug(`Received MQTT message on '${topic}' with data '${message}'`); this.eventBus.emitMQTTMessage({topic, message: message + ''}); } + + if (this.republishRetainedTimer && topic == `${settings.get().mqtt.base_topic}/bridge/state`) { + clearTimeout(this.republishRetainedTimer); + this.republishRetainedTimer = null; + } } isConnected(): boolean { diff --git a/test/controller.test.js b/test/controller.test.js index 7d0453dd6..a74963a4c 100644 --- a/test/controller.test.js +++ b/test/controller.test.js @@ -650,10 +650,21 @@ describe('Controller', () => { MQTT.publish.mockClear(); MQTT.events['connect'](); await flushPromises(); - expect(MQTT.publish).toHaveBeenCalledTimes(13); + jest.runOnlyPendingTimers(); + expect(MQTT.publish).toHaveBeenCalledTimes(12); expect(MQTT.publish).toHaveBeenCalledWith('zigbee2mqtt/bridge/info', expect.any(String), { retain: true, qos: 0 }, expect.any(Function)); }); + it('Should not republish retained messages on MQTT reconnect when retained message are sent', async () => { + await controller.start(); + MQTT.publish.mockClear(); + MQTT.events['connect'](); + await flushPromises(); + await MQTT.events.message('zigbee2mqtt/bridge/state', 'online'); + jest.runOnlyPendingTimers(); + expect(MQTT.publish).toHaveBeenCalledTimes(0); + }); + it('Should prevent any message being published with retain flag when force_disable_retain is set', async () => { settings.set(['mqtt', 'force_disable_retain'], true); await controller.mqtt.connect()