From 105a06537ce86f3d12f519cb9c782153d6e3e135 Mon Sep 17 00:00:00 2001
From: andsel <selva.andre@gmail.com>
Date: Mon, 12 Dec 2016 23:15:45 +0100
Subject: [PATCH] Moved code to make InternalPublisher.publishRetained() almost
 similar to InternalPublisher.publishStored()

---
 .../main/java/io/moquette/spi/impl/InternalRepublisher.java    | 1 -
 .../src/main/java/io/moquette/spi/impl/ProtocolProcessor.java  | 3 +++
 2 files changed, 3 insertions(+), 1 deletion(-)

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 4a4384da..1eec5548 100644
--- a/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
+++ b/broker/src/main/java/io/moquette/spi/impl/InternalRepublisher.java
@@ -44,7 +44,6 @@ class InternalRepublisher {
             clientSession.inFlightAckWaiting(pubEvt.getGuid(), pubEvt.getMessageID());
             directSend(clientSession, pubEvt.getTopic(), pubEvt.getQos(),
                     pubEvt.getMessage(), false, pubEvt.getMessageID());
-            clientSession.removeEnqueued(pubEvt.getGuid());
         }
     }
 
diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
index be638741..129bedc5 100644
--- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
@@ -413,6 +413,9 @@ public class ProtocolProcessor {
 
         LOG.info("republishing stored messages to client <{}>", clientSession.clientID);
         this.internalRepublisher.publishStored(clientSession, publishedEvents);
+        for (IMessagesStore.StoredMessage pubEvt : publishedEvents) {
+            clientSession.removeEnqueued(pubEvt.getGuid());
+        }
     }
 
     public void processPubAck(Channel channel, PubAckMessage msg) {