diff --git a/lib/mqtt.ts b/lib/mqtt.ts index d762aba4b..ed8151c20 100644 --- a/lib/mqtt.ts +++ b/lib/mqtt.ts @@ -139,14 +139,6 @@ export default class Mqtt { this.client.on("connect", this.onConnect); - this.republishRetainedTimer = setTimeout(async () => { - // Republish retained messages in case MQTT broker does not persist them. - // https://github.com/Koenkk/zigbee2mqtt/issues/9629 - for (const msg of Object.values(this.retainedMessages)) { - await this.publish(msg.topic, msg.payload, msg.options); - } - }, 2000); - // Set timer at interval to check if connected to MQTT server. this.connectionTimer = setInterval(() => { if (!this.isConnected()) { @@ -182,6 +174,17 @@ export default class Mqtt { @bind private async onConnect(): Promise { logger.info("Connected to MQTT server"); + // Armed before any await: on reconnect mqtt.js has already resubscribed, so retained + // messages (including the `bridge/info` that cancels this timer) can arrive during the publish below. + clearTimeout(this.republishRetainedTimer); + this.republishRetainedTimer = setTimeout(async () => { + // Republish retained messages in case MQTT broker does not persist them. + // https://github.com/Koenkk/zigbee2mqtt/issues/9629 + for (const msg of Object.values(this.retainedMessages)) { + await this.publish(msg.topic, msg.payload, msg.options); + } + }, 2000); + const stateData: Zigbee2MQTTAPI["bridge/state"] = {state: "online"}; await this.publish("bridge/state", JSON.stringify(stateData), {clientOptions: {retain: true, qos: 1}}); diff --git a/test/controller.test.ts b/test/controller.test.ts index a5fbcbf51..f6eeda0ec 100644 --- a/test/controller.test.ts +++ b/test/controller.test.ts @@ -1359,6 +1359,55 @@ describe("Controller", () => { expect(mockMQTTPublishAsync).toHaveBeenCalledTimes(0); }); + it("Should republish retained messages on MQTT reconnect", async () => { + await controller.start(); + await flushPromises(); + // broker kept retained messages on initial connect, no republish + await mockMQTTEvents.message("zigbee2mqtt/bridge/info", "dummy"); + + const retainedMessages = Object.keys(controller.mqtt.retainedMessages).length; + + mockMQTTPublishAsync.mockClear(); + await mockMQTTEvents.connect(); + await vi.advanceTimersByTimeAsync(2500); // before any startup configure triggers + + // bridge/state from onConnect + all retained messages + expect(mockMQTTPublishAsync).toHaveBeenCalledTimes(retainedMessages + 1); + expect(mockMQTTPublishAsync).toHaveBeenCalledWith("zigbee2mqtt/bridge/info", expect.any(String), {retain: true}); + expect(mockMQTTPublishAsync).toHaveBeenCalledWith("zigbee2mqtt/bridge/devices", expect.any(String), {retain: true}); + }); + + it("Should not republish retained messages on MQTT reconnect when retained message are sent", async () => { + await controller.start(); + await flushPromises(); + await mockMQTTEvents.message("zigbee2mqtt/bridge/info", "dummy"); + + mockMQTTPublishAsync.mockClear(); + await mockMQTTEvents.connect(); + await mockMQTTEvents.message("zigbee2mqtt/bridge/info", "dummy"); + await vi.advanceTimersByTimeAsync(2500); // before any startup configure triggers + + expect(mockMQTTPublishAsync).toHaveBeenCalledTimes(1); + expect(mockMQTTPublishAsync).toHaveBeenCalledWith("zigbee2mqtt/bridge/state", stringify({state: "online"}), {retain: true, qos: 1}); + }); + + it("Should not republish retained messages on MQTT reconnect when retained message arrives while publishing bridge state", async () => { + await controller.start(); + await flushPromises(); + await mockMQTTEvents.message("zigbee2mqtt/bridge/info", "dummy"); + + mockMQTTPublishAsync.mockClear(); + // mqtt.js resubscribes on reconnect before onConnect runs, retained messages arrive during its first publish + mockMQTTPublishAsync.mockImplementationOnce(async () => { + await mockMQTTEvents.message("zigbee2mqtt/bridge/info", "dummy"); + }); + await mockMQTTEvents.connect(); + await vi.advanceTimersByTimeAsync(2500); // before any startup configure triggers + + expect(mockMQTTPublishAsync).toHaveBeenCalledTimes(1); + expect(mockMQTTPublishAsync).toHaveBeenCalledWith("zigbee2mqtt/bridge/state", stringify({state: "online"}), {retain: true, qos: 1}); + }); + 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();