Only republish retained messages when none are received. #9629

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<void> {
@@ -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 {

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()