From 633365672cd427a24213f1e442124366375eef28 Mon Sep 17 00:00:00 2001
From: andsel <selva.andre@gmail.com>
Date: Tue, 13 Dec 2016 23:18:43 +0100
Subject: [PATCH] Changed InternalRepublisher to use the
 PersistentQueueMessageSender to send out the publish messages

---
 .../spi/impl/InternalRepublisher.java         | 60 +++++--------------
 .../moquette/spi/impl/MessagesPublisher.java  |  2 +-
 2 files changed, 17 insertions(+), 45 deletions(-)

diff --git a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
index 02d88548..db0d07b4 100644
--- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
@@ -22,9 +22,11 @@ class InternalRepublisher {
     private static final Logger LOG = LoggerFactory.getLogger(InternalRepublisher.class);
 
     private final ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors;
+    private final PersistentQueueMessageSender persistentSender;
 
     InternalRepublisher(ConcurrentMap<String, ConnectionDescriptor> connectionDescriptors) {
         this.connectionDescriptors = connectionDescriptors;
+        this.persistentSender = new PersistentQueueMessageSender(connectionDescriptors);
     }
 
     void publishRetained(ClientSession targetSession, Collection<IMessagesStore.StoredMessage> messages) {
@@ -35,7 +37,12 @@ class InternalRepublisher {
                 LOG.trace("Adding to inflight <{}>", packetID);
                 targetSession.inFlightAckWaiting(storedMsg.getGuid(), packetID);
             }
-            directSend(targetSession, storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID);
+            PublishMessage publishMsg = createPublishForQos(storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true);
+            //set the PacketIdentifier only for QoS > 0
+            if (publishMsg.getQos() != AbstractMessage.QOSType.MOST_ONE) {
+                publishMsg.setMessageID(packetID);
+            }
+            this.persistentSender.sendPublish(targetSession, publishMsg);
         }
     }
 
@@ -44,56 +51,21 @@ class InternalRepublisher {
             //put in flight zone
             LOG.trace("Adding to inflight <{}>", pubEvt.getMessageID());
             clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID());
-            directSend(clientSession, pubEvt.getTopic(), pubEvt.getQos(),
-                    pubEvt.getMessage(), false, pubEvt.getMessageID());
+            PublishMessage publishMsg = createPublishForQos(pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false);
+            //set the PacketIdentifier only for QoS > 0
+            if (publishMsg.getQos() != AbstractMessage.QOSType.MOST_ONE) {
+                publishMsg.setMessageID(pubEvt.getMessageID());
+            }
+            this.persistentSender.sendPublish(clientSession, publishMsg);
         }
     }
 
-    private void directSend(ClientSession clientsession, String topic, AbstractMessage.QOSType qos,
-                              ByteBuffer message, boolean retained, Integer messageID) {
-        String clientId = clientsession.clientID;
-        LOG.debug("directSend invoked clientId <{}> on topic <{}> QoS {} retained {} messageID {}",
-                clientId, topic, qos, retained, messageID);
+    private PublishMessage createPublishForQos(String topic, AbstractMessage.QOSType qos, ByteBuffer message, boolean retained) {
         PublishMessage pubMessage = new PublishMessage();
         pubMessage.setRetainFlag(retained);
         pubMessage.setTopicName(topic);
         pubMessage.setQos(qos);
         pubMessage.setPayload(message);
-
-        LOG.info("send publish message to <{}> on topic <{}>", clientId, topic);
-        if (LOG.isDebugEnabled()) {
-            LOG.debug("content <{}>", DebugUtils.payload2Str(message));
-        }
-        //set the PacketIdentifier only for QoS > 0
-        if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) {
-            pubMessage.setMessageID(messageID);
-        } else {
-            if (messageID != null) {
-                throw new RuntimeException("Internal bad error, trying to forwardPublish a QoS 0 message " +
-                        "with PacketIdentifier: " + messageID);
-            }
-        }
-
-        if (connectionDescriptors == null) {
-            throw new RuntimeException("Internal bad error, found connectionDescriptors to null while it should be " +
-                    "initialized, somewhere it's overwritten!!");
-        }
-        if (connectionDescriptors.get(clientId) == null) {
-            //TODO while we were publishing to the target client, that client disconnected,
-            // could happen is not an error HANDLE IT
-            throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client <%s> in cache <%s>",
-                    clientId, connectionDescriptors));
-        }
-        Channel channel = connectionDescriptors.get(clientId).channel;
-        LOG.trace("Session for clientId {}", clientId);
-        if (channel.isWritable()) {
-            LOG.debug("channel is writable");
-            //if channel is writable don't enqueue
-            channel.writeAndFlush(pubMessage);
-        } else if (pubMessage.getQos() != MOST_ONE) {
-            //enqueue to the client session
-            LOG.debug("enqueue to client session");
-            clientsession.enqueue(pubMessage);
-        }
+        return pubMessage;
     }
 }
diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
index 5c5aa4b7..40d4773f 100644
--- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
+++ b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java
@@ -52,7 +52,7 @@ class MessagesPublisher {
 
             boolean targetIsActive = this.connectionDescriptors.containsKey(sub.getClientId());
 
-            LOG.debug("Broker republishing to client <{}> topic <{}> qos <{}>, active {}",
+            LOG.debug("Broker republishing to client <{}> topicFilter <{}> qos <{}>, active {}",
                     sub.getClientId(), sub.getTopicFilter(), qos, targetIsActive);
             ByteBuffer message = origMessage.duplicate();
             if (targetIsActive) {