From 2bc5df87033287bc33d07abaeaa09096c369f921 Mon Sep 17 00:00:00 2001
From: chris-lumigrow <cschaefer@lumigrow.com>
Date: Thu, 10 Nov 2016 14:13:42 -0800
Subject: [PATCH 1/2] implement #227, use of atomicintegers on reference
 counter

---
 broker/src/main/java/io/moquette/spi/IMessagesStore.java | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java
index e2994d0d..0499579b 100644
--- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java
+++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java
@@ -21,6 +21,7 @@ import io.moquette.parser.proto.messages.AbstractMessage;
 
 import java.util.Collection;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Defines the SPI to be implemented by a StorageService that handle persistence of messages
@@ -36,7 +37,7 @@ public interface IMessagesStore {
         //Optional attribute, available only fo QoS 1 and 2
         private Integer m_msgID;
         private MessageGUID m_guid;
-        private int referenceCounter = 0;
+        private AtomicInteger referenceCounter = new AtomicInteger(0);
 
         public StoredMessage(byte[] message, AbstractMessage.QOSType qos, String topic) {
             m_qos = qos;
@@ -93,15 +94,15 @@ public interface IMessagesStore {
         }
 
         public void incReferenceCounter() {
-            this.referenceCounter++;
+            this.referenceCounter.incrementAndGet();
         }
 
         public void decReferenceCounter() {
-            this.referenceCounter--;
+            this.referenceCounter.decrementAndGet();
         }
 
         public int getReferenceCounter() {
-            return this.referenceCounter;
+            return this.referenceCounter.get();
         }
 
         @Override

From 9c62e3d05c5ab37d60423d2822c636fe161b6d9c Mon Sep 17 00:00:00 2001
From: chris-lumigrow <cschaefer@lumigrow.com>
Date: Thu, 10 Nov 2016 14:19:43 -0800
Subject: [PATCH 2/2] a bit of defensive programming in the case of negative
 reference values

---
 .../java/io/moquette/spi/persistence/MapDBMessagesStore.java | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

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 4eb497e0..f8c3d583 100644
--- a/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
+++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBMessagesStore.java
@@ -110,7 +110,10 @@ class MapDBMessagesStore implements IMessagesStore {
     void removeStoredMessage(MessageGUID guid) {
         //remove only the not retained and no more referenced
         StoredMessage storedMessage = m_persistentMessageStore.get(guid);
-        if (!storedMessage.isRetained() && storedMessage.getReferenceCounter() == 0) {
+        if (!storedMessage.isRetained() && storedMessage.getReferenceCounter() <= 0) {
+            if(storedMessage.getReferenceCounter() < 0) {
+                LOG.error("we should never have gotten a reference count less than zero");
+            }
             LOG.debug("Cleaning not retained message guid {}", guid);
             m_persistentMessageStore.remove(guid);
         }