ack

smqtt-common/src/main/java/io/github/quickmsg/common/ack/AbsAck.java

@@ -25,16 +25,16 @@ public abstract class AbsAck implements Ack {
     private final int period;
 
 
-    private final Consumer<Boolean> consumer;
+    private final Runnable cleaner;
 
 
 
-    protected AbsAck(int maxRetrySize, int period, Runnable runnable, AckManager ackManager, Consumer<Boolean> consumer) {
+    protected AbsAck(int maxRetrySize, int period, Runnable runnable, AckManager ackManager, Runnable cleaner) {
         this.maxRetrySize = maxRetrySize;
         this.period = period;
         this.runnable = runnable;
         this.ackManager = ackManager;
-        this.consumer= consumer;
+        this.cleaner= cleaner;
     }
 
     @Override
@@ -42,7 +42,6 @@ public abstract class AbsAck implements Ack {
         if (++count <= maxRetrySize+1 && !died ) {
             try {
                 log.info("task retry send ...........");
-                consumer.accept(false);
                 runnable.run();
                 ackManager.addAck(this);
             } catch (Exception e) {
@@ -51,7 +50,7 @@ public abstract class AbsAck implements Ack {
 
         }
         else {
-            consumer.accept(true);
+            cleaner.run();
         }
     }
 

smqtt-common/src/main/java/io/github/quickmsg/common/ack/RetryAck.java

 package io.github.quickmsg.common.ack;
 
-import java.util.function.Consumer;
-
 /**
  * @author luxurong
  */
@@ -11,7 +9,7 @@ public class RetryAck extends AbsAck {
     private final long id;
 
 
-    public RetryAck(long id, int maxRetrySize, int period, Runnable runnable, AckManager ackManager, Consumer<Boolean> consumer) {
+    public RetryAck(long id, int maxRetrySize, int period, Runnable runnable, AckManager ackManager, Runnable consumer) {
         super(maxRetrySize, period, runnable, ackManager,consumer);
         this.id = id;
     }

smqtt-common/src/main/java/io/github/quickmsg/common/channel/MqttChannel.java

@@ -271,18 +271,12 @@ public class MqttChannel {
                 Increase the reference count of bytebuf, and the reference count of retrybytebuf is 2
                 mqttChannel.write() method releases a reference count.
                  */
-                Runnable runnable = () -> mqttChannel.write(Mono.just(mqttMessage)).subscribe();
-                Consumer<Boolean> consumer = bool -> {
-                    if (bool) {
-                        MessageUtils.safeRelease(mqttMessage);
-                    } else {
-                        if (mqttMessage instanceof MqttPublishMessage) {
-                            ((MqttPublishMessage) mqttMessage).retain();
-                        }
-                    }
-                };
-                Ack ack = new RetryAck(mqttChannel.generateId(mqttMessage.fixedHeader().messageType(), getMessageId(mqttMessage)),
-                        5, 5, runnable, mqttChannel.getTimeAckManager(), consumer);
+                MqttMessage reply = getReplyMqttMessage(mqttMessage);
+
+                Runnable runnable = () -> mqttChannel.write(Mono.just(reply)).subscribe();
+                Runnable cleaner = () -> MessageUtils.safeRelease(reply);;
+                Ack ack = new RetryAck(mqttChannel.generateId(reply.fixedHeader().messageType(), getMessageId(reply)),
+                        5, 5, runnable, mqttChannel.getTimeAckManager(), cleaner);
                 ack.start();
                 return mqttChannel.write(Mono.just(mqttMessage)).then();
             } else {
@@ -308,7 +302,6 @@ public class MqttChannel {
             } else {
                 return mqttMessage;
             }
-
         }