Removed a queue per client of inFlightIds (never cleaned) in favor of single atomic generator, fixes issue #130

broker/src/main/java/io/moquette/spi/IMessagesStore.java

@@ -141,7 +141,5 @@ public interface IMessagesStore {
      * */
     int nextPacketID(String clientID);
 
-//    void close();
-
     void cleanRetained(String topic);
 }

broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java

@@ -23,6 +23,7 @@ import org.slf4j.LoggerFactory;
 
 import java.util.*;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * IMessagesStore implementation backed by MapDB.
@@ -36,7 +37,7 @@ class MapDBMessagesStore implements IMessagesStore {
     private DB m_db;
 
     //map clientID <-> set of currently in flight packet identifiers
-    private Map<String, Set<Integer>> m_inFlightIds;
+    private Map<String, AtomicLong> m_inFlightIds;
     //maps clientID -> guid
     private ConcurrentMap<String, String> m_retainedStore;
     //maps guid to message, it's message store
@@ -115,18 +116,19 @@ class MapDBMessagesStore implements IMessagesStore {
      * */
     @Override
     public int nextPacketID(String clientID) {
-        Set<Integer> inFlightForClient = this.m_inFlightIds.get(clientID);
-        if (inFlightForClient == null) {
-            int nextPacketId = 1;
-            inFlightForClient = new HashSet<>();
-            inFlightForClient.add(nextPacketId);
-            this.m_inFlightIds.put(clientID, inFlightForClient);
-            return nextPacketId;
+        AtomicLong atomicLong = this.m_inFlightIds.get(clientID);
+        if (atomicLong == null) {
+            AtomicLong messageId = new AtomicLong(1);
+            this.m_inFlightIds.put(clientID, messageId);
+            return 1;
         }
-        int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient);
-        int nextPacketId = (maxId + 1) % 0xFFFF;
-        inFlightForClient.add(nextPacketId);
-        return nextPacketId;
+        int incrementAndGet = (int) atomicLong.incrementAndGet();
+        if (incrementAndGet == 0XFFFF) {
+            atomicLong.set(1);
+            return 1;
+        }
+
+        return incrementAndGet;
     }
 
     @Override