Used the message store for QoS2 messages

broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java

@@ -171,4 +171,9 @@ public class ClientSession {
     public void enqueueToDeliver(String guid) {
         this.m_sessionsStore.bindToDeliver(guid, this.clientID);
     }
+
+    public PublishEvent storedMessage(int messageID) {
+        final String guid = m_sessionsStore.mapToGuid(clientID, messageID);
+        return messagesStore.getMessageByGuid(guid);
+    }
 }

broker/src/main/java/org/eclipse/moquette/spi/IMessagesStore.java

@@ -83,8 +83,10 @@ public interface IMessagesStore {
     
     void dropMessagesInSession(String clientID);
 
+    PublishEvent getMessageByGuid(String guid);
+
     /**
-     * Return the next valid packetIdentifer for the given client session.
+     * Return the next valid packetIdentifier for the given client session.
      * */
     int nextPacketID(String clientID);
 

broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java

@@ -91,4 +91,6 @@ public interface ISessionsStore {
     void secondPhaseAcknowledged(String clientID, int messageID);
 
     void secondPhaseAckWaiting(String clientID, int messageID);
+
+    String mapToGuid(String clientID, int messageID);
 }

broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java

@@ -121,7 +121,6 @@ public class ProtocolProcessor {
               ISessionsStore sessionsStore,
               IAuthenticator authenticator,
               boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) {
-        //m_clientIDs = clientIDs;
         this.m_clientIDs = new ConcurrentHashMap<>();
         this.m_interceptor = interceptor;
         this.subscriptions = subscriptions;
@@ -274,16 +273,10 @@ public class ProtocolProcessor {
         final String topic = msg.getTopicName();
         //check if the topic can be wrote
         String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME);
-        if (m_authorizator.canWrite(topic, user, clientID)) {
-            executePublish(clientID, msg);
-            m_interceptor.notifyTopicPublished(msg, clientID);
-        } else {
+        if (!m_authorizator.canWrite(topic, user, clientID)) {
             LOG.debug("topic {} doesn't have write credentials", topic);
+            return;
         }
-    }
-        
-    private void executePublish(String clientID, PublishMessage msg) {
-        final String topic = msg.getTopicName();
         final AbstractMessage.QOSType qos = msg.getQos();
         final ByteBuffer message = msg.getPayload();
         boolean retain = msg.isRetainFlag();
@@ -298,9 +291,11 @@ public class ProtocolProcessor {
             sendPubAck(new PubAckEvent(messageID, clientID));
             LOG.debug("replying with PubAck to MSG ID {}", messageID);
         }  else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2
-            String publishKey = String.format("%s%d", clientID, messageID);
+//            String publishKey = String.format("%s%d", clientID, messageID);
             //store the message in temp store
-            m_messagesStore.persistQoS2Message(publishKey, publishEvt);
+//TODO use the global message store
+//            m_messagesStore.persistQoS2Message(publishKey, publishEvt);
+            m_messagesStore.storePublishForFuture(publishEvt);
             sendPubRec(clientID, messageID);
             //Next the client will send us a pub rel
             //NB publish to subscribers for QoS 2 happen upon PUBREL from publisher
@@ -308,14 +303,15 @@ public class ProtocolProcessor {
 
         if (retain) {
             if (qos == AbstractMessage.QOSType.MOST_ONE) {
-                //QoS == 0 && retain => clean old retained 
+                //QoS == 0 && retain => clean old retained
                 m_messagesStore.cleanRetained(topic);
             } else {
                 m_messagesStore.storeRetained(topic, message, qos);
             }
         }
+        m_interceptor.notifyTopicPublished(msg, clientID);
     }
-    
+        
     /**
      * Specialized version to publish will testament message.
      */
@@ -460,10 +456,13 @@ public class ProtocolProcessor {
         String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
         int messageID = msg.getMessageID();
         LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
-        String publishKey = String.format("%s%d", clientID, messageID);
-        PublishEvent evt = m_messagesStore.retrieveQoS2Message(publishKey);
+//        String publishKey = String.format("%s%d", clientID, messageID);
+//        PublishEvent evt = m_messagesStore.retrieveQoS2Message(publishKey);
+        //get guid, get message
+        ClientSession targetSession = m_sessionsStore.sessionForClient(clientID);
+        PublishEvent evt = targetSession.storedMessage(messageID);
         forward2Subscribers(evt);
-        m_messagesStore.removeQoS2Message(publishKey);
+//        m_messagesStore.removeQoS2Message(publishKey);
 
         if (evt.isRetain()) {
             final String topic = evt.getTopic();

broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java

@@ -170,6 +170,8 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
         String guid = UUID.randomUUID().toString();
         evt.setGuid(guid);
         m_persistentMessageStore.put(guid, convertToStored(evt));
+        ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap("guidsMapping" + evt.getClientID());
+        messageIdToGuid.put(evt.getMessageID(), guid);
         return guid;
     }
 
@@ -182,10 +184,23 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
         return ret;
     }
 
+    @Override
     public void dropMessagesInSession(String clientID) {
+        m_db.getHashMap("guidsMapping" + clientID).clear();
         m_persistentMessageStore.remove(clientID);
     }
 
+    @Override
+    public PublishEvent getMessageByGuid(String guid) {
+        return convertFromStored(m_persistentMessageStore.get(guid));
+    }
+
+    @Override
+    public String mapToGuid(String clientID, int messageID) {
+        ConcurrentMap<Integer, String> messageIdToGuid = m_db.getHashMap("guidsMapping" + clientID);
+        return messageIdToGuid.get(messageID);
+    }
+
     /**
      * Return the next valid packetIdentifier for the given client session.
      * */