From 2cd60772874ce89cc6a2136168c68d6c5d6cb137 Mon Sep 17 00:00:00 2001
From: andsel <selva.andre@gmail.com>
Date: Sun, 15 Nov 2015 12:54:30 +0100
Subject: [PATCH] Fixed packetId missed cleanup

---
 .../spi/persistence/MapDBPersistentStore.java | 65 ++++---------------
 .../persistence/MapDBPersistentStoreTest.java |  5 +-
 2 files changed, 14 insertions(+), 56 deletions(-)

diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java
index 46feca80..43928079 100644
--- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java
+++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java
@@ -76,8 +76,8 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
     private ConcurrentMap<String, Set<Subscription>> m_persistentSubscriptions;
     private ConcurrentMap<String, PersistentSession> m_persistentSessions;
 
-    //maps clientID->[guid*]
-    private ConcurrentMap<String, Set<String>> m_enqueuedStore;
+    //maps clientID->[guid*], insertion order cares, it's queue
+    private ConcurrentMap<String, List<String>> m_enqueuedStore;
     //maps clientID->[messageID*]
     private ConcurrentMap<String, Set<Integer>> m_secondPhaseStore;
 
@@ -182,59 +182,10 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
         return ret;
     }
 
-//    @Override
-//    public List<PublishEvent> listMessagesInSession(String clientID) {
-//        List<PublishEvent> liveEvts = new ArrayList<>();
-//        List<StoredPublishEvent> storedEvts = defaultGet(m_persistentMessageStore, clientID, Collections.<StoredPublishEvent>emptyList());
-//
-//        for (StoredPublishEvent storedEvt : storedEvts) {
-//            liveEvts.add(convertFromStored(storedEvt));
-//        }
-//        return liveEvts;
-//    }
-//
-//    @Override
-//    public void removeMessage(String clientID, Integer messageID) {
-//        List<StoredPublishEvent> events = m_persistentMessageStore.get(clientID);
-//        if (events == null) {
-//            return;
-//        }
-//        StoredPublishEvent toRemoveEvt = null;
-//        for (StoredPublishEvent evt : events) {
-//            if (evt.getMessageID() == null && messageID == null) {
-//                //was a qos0 message (no ID)
-//                toRemoveEvt = evt;
-//            }
-//            if (evt.getMessageID().equals(messageID)) {
-//                toRemoveEvt = evt;
-//            }
-//        }
-//        events.remove(toRemoveEvt);
-//        m_persistentMessageStore.put(clientID, events);
-//    }
-
     public void dropMessagesInSession(String clientID) {
         m_persistentMessageStore.remove(clientID);
     }
 
-    //----------------- In flight methods -----------------
-//    @Override
-//    public void cleanTemporaryPublish(String clientID, int packetID) {
-//        String publishKey = String.format("%s%d", clientID, packetID);
-//        m_inflightStore.remove(publishKey);
-//        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
-//        if (inFlightForClient != null) {
-//            inFlightForClient.remove(packetID);
-//        }
-//    }
-//
-//    @Override
-//    public void storeTemporaryPublish(PublishEvent evt, String clientID, int packetID) {
-//        String publishKey = String.format("%s%d", clientID, packetID);
-//        StoredPublishEvent storedEvt = convertToStored(evt);
-//        m_inflightStore.put(publishKey, storedEvt);
-//    }
-
     /**
      * Return the next valid packetIdentifier for the given client session.
      * */
@@ -377,6 +328,12 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
             return;
         }
         m.remove(messageID);
+
+        //remove from the ids store
+        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
+        if (inFlightForClient != null) {
+            inFlightForClient.remove(messageID);
+        }
     }
 
     @Override
@@ -391,19 +348,19 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
 
     @Override
     public void bindToDeliver(String guid, String clientID) {
-        Set<String> guids = defaultGet(m_enqueuedStore, clientID, new HashSet<String>());
+        List<String> guids = defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
         guids.add(guid);
         m_enqueuedStore.put(clientID, guids);
     }
 
     @Override
     public Collection<String> enqueued(String clientID) {
-        return defaultGet(m_enqueuedStore, clientID, new HashSet<String>());
+        return defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
     }
 
     @Override
     public void removeEnqueued(String clientID, String guid) {
-        Set<String> guids = defaultGet(m_enqueuedStore, clientID, new HashSet<String>());
+        List<String> guids = defaultGet(m_enqueuedStore, clientID, new ArrayList<String>());
         guids.remove(guid);
         m_enqueuedStore.put(clientID, guids);
     }
diff --git a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java
index 189518f9..1cfd619f 100644
--- a/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java
+++ b/broker/src/test/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStoreTest.java
@@ -22,7 +22,6 @@ import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
-import java.io.File;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
 
@@ -87,11 +86,13 @@ public class MapDBPersistentStoreTest {
     @Test
     public void testNextPacketID() {
         //request a first ID
+
         int packetId = m_storageService.nextPacketID("CLIENT");
+        m_storageService.inFlight("CLIENT", packetId, "ABCDE"); //simulate an inflight
         assertEquals(1, packetId);
 
         //release the ID
-//        m_storageService.cleanTemporaryPublish("CLIENT", packetId);
+        m_storageService.inFlightAck("CLIENT", packetId);
 
         //request a second packetID, counter restarts from 0
         packetId = m_storageService.nextPacketID("CLIENT");