remove stored subsCalled when connection is lost

client.go

@@ -202,6 +202,19 @@ func (c *Client) attemptSingleConnection(addrs []TCPAddress) error {
 	return c.resumeSubscriptions()
 }
 
+func (c *Client) removeStoreSubscriptions(cc mqtt.Client) {
+	c.clientMu.RLock()
+	defer c.clientMu.RUnlock()
+
+	for _, v := range c.mqttClients {
+		if v.client == cc {
+			v.mu.Lock()
+			v.subsCalled.Delete(v.subsCalled.Values()...)
+			v.mu.Unlock()
+		}
+	}
+}
+
 func toClientOptions(c *Client, o *clientOptions, idSuffix string) *mqtt.ClientOptions {
 	opts := mqtt.NewClientOptions()
 
@@ -228,7 +241,7 @@ func toClientOptions(c *Client, o *clientOptions, idSuffix string) *mqtt.ClientO
 		SetConnectTimeout(o.connectTimeout).
 		SetMaxReconnectInterval(o.maxReconnectInterval).
 		SetReconnectingHandler(reconnectHandler(c, o)).
-		SetConnectionLostHandler(connectionLostHandler(o)).
+		SetConnectionLostHandler(connectionLostHandler(c, o)).
 		SetOnConnectHandler(onConnectHandler(c, o))
 
 	return opts
@@ -283,7 +296,7 @@ func reconnectHandler(client PubSub, o *clientOptions) mqtt.ReconnectHandler {
 	}
 }
 
-func connectionLostHandler(o *clientOptions) mqtt.ConnectionLostHandler {
+func connectionLostHandler(c *Client, o *clientOptions) mqtt.ConnectionLostHandler {
 	return func(cc mqtt.Client, err error) {
 		if o.logger != nil {
 			o.logger.Error(context.Background(), err, map[string]any{
@@ -292,6 +305,8 @@ func connectionLostHandler(o *clientOptions) mqtt.ConnectionLostHandler {
 			})
 		}
 
+		c.removeStoreSubscriptions(cc)
+
 		if o.onConnectionLostHandler != nil {
 			o.onConnectionLostHandler(err)
 		}

client_test.go

@@ -335,7 +335,7 @@ func Test_connectionLostHandler(t *testing.T) {
 		"client_id": "clientID",
 	}).Return()
 
-	f := connectionLostHandler(c.options)
+	f := connectionLostHandler(c, c.options)
 	f(c.mqttClient, errors.New("disconnected"))
 	ml.AssertExpectations(t)
 }