Removed the necessity to use BestEffortMessageSender

broker/src/main/java/io/moquette/spi/impl/BestEffortMessageSender.java

@@ -48,12 +48,5 @@ class BestEffortMessageSender {
         }
     }
 
-    static PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message) {
-        PublishMessage pubMessage = new PublishMessage();
-        pubMessage.setRetainFlag(false);
-        pubMessage.setTopicName(topic);
-        pubMessage.setQos(qos);
-        pubMessage.setPayload(message);
-        return pubMessage;
-    }
+
 }

broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java

@@ -15,6 +15,8 @@ import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.ConcurrentMap;
 
+import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE;
+
 class InternalRepublisher {
 
     private static final Logger LOG = LoggerFactory.getLogger(InternalRepublisher.class);
@@ -88,7 +90,7 @@ class InternalRepublisher {
             LOG.debug("channel is writable");
             //if channel is writable don't enqueue
             channel.writeAndFlush(pubMessage);
-        } else {
+        } else if (pubMessage.getQos() != MOST_ONE) {
             //enqueue to the client session
             LOG.debug("enqueue to client session");
             clientsession.enqueue(pubMessage);

broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java

@@ -15,8 +15,7 @@ import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.concurrent.ConcurrentMap;
 
-import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE;
-import static io.moquette.spi.impl.BestEffortMessageSender.createPublishForQos;
+import static io.moquette.spi.impl.PersistentQueueMessageSender.createPublishForQos;
 import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired;
 
 class MessagesPublisher {
@@ -25,7 +24,6 @@ class MessagesPublisher {
     private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
     private final ISessionsStore m_sessionsStore;
     private final IMessagesStore m_messagesStore;
-    private final BestEffortMessageSender bestEffortSender;
     private final PersistentQueueMessageSender persistentSender;
 
     public MessagesPublisher(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors, ISessionsStore sessionsStore,
@@ -33,7 +31,6 @@ class MessagesPublisher {
         this.connectionDescriptors = connectionDescriptors;
         this.m_sessionsStore = sessionsStore;
         this.m_messagesStore = messagesStore;
-        this.bestEffortSender = new BestEffortMessageSender(connectionDescriptors);
         this.persistentSender = new PersistentQueueMessageSender(connectionDescriptors);
     }
 
@@ -61,8 +58,8 @@ class MessagesPublisher {
             if (targetIsActive) {
                 //QoS 0
                 if (qos == AbstractMessage.QOSType.MOST_ONE) {
-                    PublishMessage publishMsg = BestEffortMessageSender.createPublishForQos(topic, MOST_ONE, message);
-                    this.bestEffortSender.publishQos0(targetSession, publishMsg);
+                    PublishMessage publishMsg = createPublishForQos(topic, qos, message);
+                    this.persistentSender.publishQos2(targetSession, publishMsg);
                 } else {
                     //QoS 1 or 2
                     int messageId = targetSession.nextPacketId();

broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java

@@ -1,5 +1,6 @@
 package io.moquette.spi.impl;
 
+import io.moquette.parser.proto.messages.AbstractMessage;
 import io.moquette.parser.proto.messages.PublishMessage;
 import io.moquette.server.ConnectionDescriptor;
 import io.moquette.spi.ClientSession;
@@ -7,8 +8,11 @@ import io.netty.channel.Channel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.nio.ByteBuffer;
 import java.util.concurrent.ConcurrentMap;
 
+import static io.moquette.parser.proto.messages.AbstractMessage.QOSType.MOST_ONE;
+
 class PersistentQueueMessageSender {
 
     private static final Logger LOG = LoggerFactory.getLogger(PersistentQueueMessageSender.class);
@@ -44,11 +48,20 @@ class PersistentQueueMessageSender {
             LOG.debug("channel is writable");
             //if channel is writable don't enqueue
             channel.writeAndFlush(pubMessage);
-        } else {
+        } else if (pubMessage.getQos() != MOST_ONE) {
             //enqueue to the client session
             LOG.debug("enqueue to client session");
             clientsession.enqueue(pubMessage);
         }
     }
 
+    static PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message) {
+        PublishMessage pubMessage = new PublishMessage();
+        pubMessage.setRetainFlag(false);
+        pubMessage.setTopicName(topic);
+        pubMessage.setQos(qos);
+        pubMessage.setPayload(message);
+        return pubMessage;
+    }
+
 }