Moved code to make InternalPublisher.publishRetained() almost similar to InternalPublisher.publishStored()

2016-12-13 23:43:02 +01:00
parent 5d02239dc1
commit 105a06537c
2 changed files with 3 additions and 1 deletions
@@ -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());
        }
    }

@@ -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) {