Fixed NPE exception of inflight ACK because the publisher with clean=true removed all messages, also the published from message store

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java

@@ -43,6 +43,7 @@ import io.netty.channel.ChannelPipeline;
 import io.netty.handler.timeout.IdleStateHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import sun.misc.MessageUtils;
 
 /**
  * Class responsible to handle the logic of MQTT protocol it's the director of
@@ -424,13 +425,16 @@ public class ProtocolProcessor {
         String username = NettyUtils.userName(channel);
         LOG.trace("retrieving inflight for messageID <{}>", messageID);
 
-        //Remove the message from message store
         ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
         StoredMessage inflightMsg = targetSession.getInflightMessage(messageID);
         targetSession.inFlightAcknowledged(messageID);
 
         String topic = inflightMsg.getTopic();
 
+        MessageGUID guid = inflightMsg.getGuid();
+        //Remove the message from message store
+        m_messagesStore.decUsageCounter(guid);
+
         m_interceptor.notifyMessageAcknowledged(new InterceptAcknowledgedMessage(inflightMsg, topic, username));
     }
 
@@ -632,9 +636,12 @@ public class ProtocolProcessor {
         }
 
         if (descriptor.cleanSession) {
-            LOG.debug("Removing messages in session for client <{}>", clientID);
-            this.m_messagesStore.dropMessagesInSession(clientID);
-            LOG.debug("Removed messages in session for client <{}>", clientID);
+            LOG.debug("Removing messages in session's queue for client <{}>", clientID);
+            Collection<MessageGUID> queue = this.m_sessionsStore.enqueued(clientID);
+            for (MessageGUID guid : queue) {
+                this.m_sessionsStore.removeEnqueued(clientID, guid);
+            }
+            LOG.debug("Removed messages in session for client's queue <{}>", clientID);
         }
         return true;
     }

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

@@ -227,9 +227,6 @@ class MapDBSessionsStore implements ISessionsStore {
         List<MessageGUID> guids = Utils.defaultGet(m_enqueuedStore, clientID, new ArrayList<MessageGUID>());
         guids.remove(guid);
         m_enqueuedStore.put(clientID, guids);
-        m_messagesStore.decUsageCounter(guid);
-        //if counter gets to 0 then remove from storage
-        m_messagesStore.removeStoredMessage(guid);
     }
 
     @Override