Republish retained messages when reconnected to MQTT server. #9629

lib/mqtt.ts

@@ -10,6 +10,8 @@ export default class MQTT {
     private connectionTimer: NodeJS.Timeout;
     private client: mqtt.MqttClient;
     private eventBus: EventBus;
+    private retainedMessages: {[s: string]: {payload: string, options: MQTTOptions,
+        skipLog: boolean, skipReceive: boolean, topic: string, base: string}} = {};
 
     constructor(eventBus: EventBus) {
         this.eventBus = eventBus;
@@ -94,6 +96,12 @@ export default class MQTT {
         }, utils.seconds(10));
 
         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));
+
         this.subscribe(`${settings.get().mqtt.base_topic}/#`);
         await this.publishStateOnline();
     }
@@ -137,6 +145,15 @@ export default class MQTT {
             this.publishedTopics.add(topic);
         }
 
+        if (options.retain) {
+            if (payload) {
+                this.retainedMessages[topic] =
+                    {payload, options, skipReceive, skipLog, topic: topic.substring(base.length + 1), base};
+            } else {
+                delete this.retainedMessages[topic];
+            }
+        }
+
         this.eventBus.emitMQTTMessagePublished({topic, payload, options: {...defaultOptions, ...options}});
 
         if (!this.isConnected()) {

test/controller.test.js

@@ -645,6 +645,15 @@ describe('Controller', () => {
         expect(MQTT.connect).toHaveBeenCalledWith("mqtt://localhost", expected);
     });
 
+    it('Should republish retained messages on MQTT reconnect', async () => {
+        await controller.start();
+        MQTT.publish.mockClear();
+        MQTT.events['connect']();
+        await flushPromises();
+        expect(MQTT.publish).toHaveBeenCalledTimes(13);
+        expect(MQTT.publish).toHaveBeenCalledWith('zigbee2mqtt/bridge/info', expect.any(String), { retain: true, qos: 0 }, expect.any(Function));
+    });
+
     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()