From 2839f873ed9ca4ff29c25acea87ab8de2933ef16 Mon Sep 17 00:00:00 2001 From: andig Date: Sat, 23 May 2026 10:21:45 +0200 Subject: [PATCH] MQTT: serialize resubscribe after reconnect (#30138) --- plugin/mqtt/client.go | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/plugin/mqtt/client.go b/plugin/mqtt/client.go index 73da06bae..123e5f39a 100644 --- a/plugin/mqtt/client.go +++ b/plugin/mqtt/client.go @@ -135,12 +135,24 @@ func (m *Client) ConnectionHandler(client paho.Client) { m.log.DEBUG.Printf("%s connected", m.broker) m.mux.Lock() - defer m.mux.Unlock() - + topics := make([]string, 0, len(m.listener)) for topic := range m.listener { - m.log.DEBUG.Printf("%s subscribe %s", m.broker, topic) - go m.listen(topic) + topics = append(topics, topic) } + m.mux.Unlock() + + // Resubscribe sequentially to avoid bursting the broker right after reconnect. + go func() { + for _, topic := range topics { + m.log.DEBUG.Printf("%s subscribe %s", m.broker, topic) + token := m.listen(topic) + if !token.WaitTimeout(request.Timeout) { + m.log.ERROR.Printf("subscribe %s: timeout", topic) + } else if err := token.Error(); err != nil { + m.log.ERROR.Printf("subscribe %s: %v", topic, err) + } + } + }() } // Cleanup recursively removes a topic