Fix duplication of MQTT packets in MQTT Client

netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttClientImpl.java

@@ -365,7 +365,8 @@ final class MqttClientImpl implements MqttClient {
         MqttFixedHeader fixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, qos, retain, 0);
         MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(topic, getNewMessageId().messageId());
         MqttPublishMessage message = new MqttPublishMessage(fixedHeader, variableHeader, payload);
-        MqttPendingPublish pendingPublish = new MqttPendingPublish(variableHeader.packetId(), future, payload.retain(), message, qos);
+        MqttPendingPublish pendingPublish = new MqttPendingPublish(variableHeader.packetId(), future,
+                payload.retain(), message, qos, () -> !pendingPublishes.containsKey(variableHeader.packetId()));
         this.pendingPublishes.put(pendingPublish.getMessageId(), pendingPublish);
         ChannelFuture channelFuture = this.sendAndFlushPacket(message);
 
@@ -471,7 +472,8 @@ final class MqttClientImpl implements MqttClient {
         MqttSubscribePayload payload = new MqttSubscribePayload(Collections.singletonList(subscription));
         MqttSubscribeMessage message = new MqttSubscribeMessage(fixedHeader, variableHeader, payload);
 
-        final MqttPendingSubscription pendingSubscription = new MqttPendingSubscription(future, topic, message);
+        final MqttPendingSubscription pendingSubscription = new MqttPendingSubscription(future, topic, message,
+                () -> !pendingSubscriptions.containsKey(variableHeader.messageId()));
         pendingSubscription.addHandler(handler, once);
         this.pendingSubscriptions.put(variableHeader.messageId(), pendingSubscription);
         this.pendingSubscribeTopics.add(topic);
@@ -489,7 +491,8 @@ final class MqttClientImpl implements MqttClient {
             MqttUnsubscribePayload payload = new MqttUnsubscribePayload(Collections.singletonList(topic));
             MqttUnsubscribeMessage message = new MqttUnsubscribeMessage(fixedHeader, variableHeader, payload);
 
-            MqttPendingUnsubscription pendingUnsubscription = new MqttPendingUnsubscription(promise, topic, message);
+            MqttPendingUnsubscription pendingUnsubscription = new MqttPendingUnsubscription(promise, topic, message,
+                    () -> !pendingServerUnsubscribes.containsKey(variableHeader.messageId()));
             this.pendingServerUnsubscribes.put(variableHeader.messageId(), pendingUnsubscription);
             pendingUnsubscription.startRetransmissionTimer(this.eventLoop.next(), this::sendAndFlushPacket);
 

netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingPublish.java

@@ -32,19 +32,21 @@ final class MqttPendingPublish{
     private final MqttPublishMessage message;
     private final MqttQoS qos;
 
-    private final RetransmissionHandler<MqttPublishMessage> publishRetransmissionHandler = new RetransmissionHandler<>();
+    private final RetransmissionHandler<MqttPublishMessage> publishRetransmissionHandler;
-    private final RetransmissionHandler<MqttMessage> pubrelRetransmissionHandler = new RetransmissionHandler<>();
+    private final RetransmissionHandler<MqttMessage> pubrelRetransmissionHandler;
 
     private boolean sent = false;
 
-    MqttPendingPublish(int messageId, Promise<Void> future, ByteBuf payload, MqttPublishMessage message, MqttQoS qos) {
+    MqttPendingPublish(int messageId, Promise<Void> future, ByteBuf payload, MqttPublishMessage message, MqttQoS qos, PendingOperation operation) {
         this.messageId = messageId;
         this.future = future;
         this.payload = payload;
         this.message = message;
         this.qos = qos;
 
+        this.publishRetransmissionHandler = new RetransmissionHandler<>(operation);
         this.publishRetransmissionHandler.setOriginalMessage(message);
+        this.pubrelRetransmissionHandler = new RetransmissionHandler<>(operation);
     }
 
     int getMessageId() {
@@ -102,5 +104,8 @@ final class MqttPendingPublish{
     void onChannelClosed() {
         this.publishRetransmissionHandler.stop();
         this.pubrelRetransmissionHandler.stop();
+        if (payload != null) {
+            payload.release();
+        }
     }
 }

netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingSubscription.java

@@ -30,15 +30,16 @@ final class MqttPendingSubscription{
     private final Set<MqttPendingHandler> handlers = new HashSet<>();
     private final MqttSubscribeMessage subscribeMessage;
 
-    private final RetransmissionHandler<MqttSubscribeMessage> retransmissionHandler = new RetransmissionHandler<>();
+    private final RetransmissionHandler<MqttSubscribeMessage> retransmissionHandler;
 
     private boolean sent = false;
 
-    MqttPendingSubscription(Promise<Void> future, String topic, MqttSubscribeMessage message) {
+    MqttPendingSubscription(Promise<Void> future, String topic, MqttSubscribeMessage message, PendingOperation operation) {
         this.future = future;
         this.topic = topic;
         this.subscribeMessage = message;
 
+        this.retransmissionHandler = new RetransmissionHandler<>(operation);
         this.retransmissionHandler.setOriginalMessage(message);
     }
 

netty-mqtt/src/main/java/org/thingsboard/mqtt/MqttPendingUnsubscription.java

@@ -26,12 +26,13 @@ final class MqttPendingUnsubscription{
     private final Promise<Void> future;
     private final String topic;
 
-    private final RetransmissionHandler<MqttUnsubscribeMessage> retransmissionHandler = new RetransmissionHandler<>();
+    private final RetransmissionHandler<MqttUnsubscribeMessage> retransmissionHandler;
 
-    MqttPendingUnsubscription(Promise<Void> future, String topic, MqttUnsubscribeMessage unsubscribeMessage) {
+    MqttPendingUnsubscription(Promise<Void> future, String topic, MqttUnsubscribeMessage unsubscribeMessage, PendingOperation operation) {
         this.future = future;
         this.topic = topic;
 
+        this.retransmissionHandler = new RetransmissionHandler<>(operation);
         this.retransmissionHandler.setOriginalMessage(unsubscribeMessage);
     }
 

netty-mqtt/src/main/java/org/thingsboard/mqtt/PendingOperation.java

@@ -0,0 +1,22 @@
+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.mqtt;
+
+public interface PendingOperation {
+
+    boolean isCanceled();
+
+}

netty-mqtt/src/main/java/org/thingsboard/mqtt/RetransmissionHandler.java

@@ -21,12 +21,16 @@ import io.netty.handler.codec.mqtt.MqttMessage;
 import io.netty.handler.codec.mqtt.MqttMessageType;
 import io.netty.handler.codec.mqtt.MqttQoS;
 import io.netty.util.concurrent.ScheduledFuture;
+import lombok.RequiredArgsConstructor;
 
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 
+@RequiredArgsConstructor
 final class RetransmissionHandler<T extends MqttMessage> {
 
+    private volatile boolean stopped;
+    private final PendingOperation pendingOperation;
     private ScheduledFuture<?> timer;
     private int timeout = 10;
     private BiConsumer<MqttFixedHeader, T> handler;
@@ -44,7 +48,13 @@ final class RetransmissionHandler<T extends MqttMessage> {
     }
 
     private void startTimer(EventLoop eventLoop) {
+        if (stopped || pendingOperation.isCanceled()) {
+            return;
+        }
         this.timer = eventLoop.schedule(() -> {
+            if (stopped || pendingOperation.isCanceled()) {
+                return;
+            }
             this.timeout += 5;
             boolean isDup = this.originalMessage.fixedHeader().isDup();
             if (this.originalMessage.fixedHeader().messageType() == MqttMessageType.PUBLISH && this.originalMessage.fixedHeader().qosLevel() != MqttQoS.AT_MOST_ONCE) {
@@ -57,6 +67,7 @@ final class RetransmissionHandler<T extends MqttMessage> {
     }
 
     void stop() {
+        stopped = true;
         if (this.timer != null) {
             this.timer.cancel(true);
         }