diff --git a/lib/mqtt.ts b/lib/mqtt.ts index e9182043e..9a3fefd81 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 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()) { diff --git a/test/controller.test.js b/test/controller.test.js index de3e65e4f..7d0453dd6 100644 --- a/test/controller.test.js +++ b/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()