diff --git a/ChangeLog.txt b/ChangeLog.txt index 4a858d23..a8a3f9ba 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -18,6 +18,7 @@ Version 0.8: Added internal publish once the broker is embedded (issue #113) Removed client active/not active state from persistence (issue #118) Avoid deletion of client descriptor in case of multiple clients with same clientID (issues #116 and #124) + Removed a queue per client of inFlightIds (never cleaned) in favor of single atomic generator Version 0.7: Fix issue #38, added aligned to MQTT 3.1.1 diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index 8d76f238..8b2d1188 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/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); } diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java index b4785dc4..f8ab8a34 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java +++ b/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> m_inFlightIds; + private Map m_inFlightIds; //maps clientID -> guid private ConcurrentMap 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 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 diff --git a/pom.xml b/pom.xml index ba1c11b1..2ff1c496 100644 --- a/pom.xml +++ b/pom.xml @@ -3,7 +3,7 @@ 4.0.0 UTF-8 - 4.0.30.Final + 4.0.33.Final 1.7 1.7 andsel/maven