Refactored ProtocolProcessor.processSubscribe to use the ClientSession concept instead of directe access to map storage concepts

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

@@ -15,22 +15,36 @@
  */
 package org.eclipse.moquette.spi;
 
+import org.eclipse.moquette.proto.messages.AbstractMessage;
 import org.eclipse.moquette.spi.impl.events.PublishEvent;
+import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
+import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 /**
  * @author andrea
  */
 public class ClientSession {
 
+    private final static Logger LOG = LoggerFactory.getLogger(ClientSession.class);
+
     public final String clientID;
 
     private final IMessagesStore messagesStore;
 
-    public ClientSession(String clientID, IMessagesStore messagesStore) {
+    private final ISessionsStore m_sessionsStore;
+
+    private Set<Subscription> subscriptions = new HashSet<>();
+
+    public ClientSession(String clientID, IMessagesStore messagesStore, ISessionsStore sessionsStore) {
         this.clientID = clientID;
         this.messagesStore = messagesStore;
+        this.m_sessionsStore = sessionsStore;
     }
     //List of client's subscriptions
     //list of messages not acknowledged by client
@@ -56,4 +70,18 @@ public class ClientSession {
                 "clientID='" + clientID + '\'' +
                 '}';
     }
+
+    public boolean subscribe(String topicFilter, Subscription newSubscription) {
+        LOG.info("<{}> subscribed to topicFilter <{}> with QoS {}",
+                newSubscription.getClientId(), topicFilter,
+                AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
+        boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter());
+        if (!validTopic) {
+            //send SUBACK with 0x80 for this topic filter
+            return false;
+        }
+        subscriptions.add(newSubscription);
+        m_sessionsStore.addNewSubscription(newSubscription);
+        return true;
+    }
 }

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

@@ -465,9 +465,8 @@ public class ProtocolProcessor {
             if (m_clientIDs.get(clientId) == null) {
                 throw new RuntimeException(String.format("Can't find a ConnectionDescriptor for client %s in cache %s", clientId, m_clientIDs));
             }
-//            LOG.debug("Session for clientId " + clientId + " is " + m_clientIDs.get(clientId).getSession());
             m_clientIDs.get(clientId).getSession().write(pubAckMessage);
-        }catch(Throwable t) {
+        } catch(Throwable t) {
             LOG.error(null, t);
         }
     }
@@ -531,7 +530,6 @@ public class ProtocolProcessor {
             //cleanup topic subscriptions
             cleanSession(clientID);
         }
-//        m_notifier.disconnect(evt.getSession());
         m_clientIDs.remove(clientID);
         session.close(true);
 
@@ -596,35 +594,36 @@ public class ProtocolProcessor {
         boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION);
         LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID());
 
+        ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
         //ack the client
         SubAckMessage ackMessage = new SubAckMessage();
         ackMessage.setMessageID(msg.getMessageID());
 
+        List<Subscription> newSubscriptions = new ArrayList<>();
         for (SubscribeMessage.Couple req : msg.subscriptions()) {
             AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.getQos());
             Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos, cleanSession);
-            boolean valid = subscribeSingleTopic(newSubscription, req.getTopicFilter());
+            //boolean valid = subscribeSingleTopic(newSubscription, req.getTopicFilter());
+            boolean valid = clientSession.subscribe(req.getTopicFilter(), newSubscription);
             ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE);
+            if (valid) {
+                newSubscriptions.add(newSubscription);
+            }
         }
 
+        //save session, persist subscriptions from session
         LOG.debug("SUBACK for packetID {}", msg.getMessageID());
         session.write(ackMessage);
+
+        //fire the publish
+        for(Subscription subscription : newSubscriptions) {
+            subscribeSingleTopic(subscription, subscription.getTopicFilter());
+        }
     }
     
     private boolean subscribeSingleTopic(Subscription newSubscription, final String topic) {
-        LOG.info("<{}> subscribed to topic <{}> with QoS {}", 
-                newSubscription.getClientId(), topic, 
-                AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
-        boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter());
-        if (!validTopic) {
-            //send SUBACK with 0x80 for this topic filter
-            return false;
-        }
         subscriptions.add(newSubscription);
 
-        //notify the Observables
-        m_interceptor.notifyTopicSubscribed(newSubscription);
-
         //scans retained messages to be published to the new subscription
         Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
             public boolean match(String key) {
@@ -640,6 +639,9 @@ public class ProtocolProcessor {
                     m_messagesStore.nextPacketID(newSubscription.getClientId());
             sendPublish(newSubscription.getClientId(), storedMsg.getTopic(), storedMsg.getQos(), storedMsg.getPayload(), true, packetID);
         }
+
+        //notify the Observables
+        m_interceptor.notifyTopicSubscribed(newSubscription);
         return true;
     }
 }

broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java

@@ -20,6 +20,7 @@ import java.util.*;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.atomic.AtomicReference;
 
+import org.eclipse.moquette.spi.ClientSession;
 import org.eclipse.moquette.spi.ISessionsStore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -163,8 +164,14 @@ public class SubscriptionsStore {
         return new NodeCouple(newRoot, current);
     }
 
+//    public void save(ClientSession clientSession) {
+//        for(Subscription subscription : clientSession.getSubscriptions()) {
+//            add(subscription);
+//        }
+//    }
+
     public void add(Subscription newSubscription) {
-        m_sessionsStore.addNewSubscription(newSubscription);
+//        m_sessionsStore.addNewSubscription(newSubscription);
         addDirect(newSubscription);
     }
 

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

@@ -309,7 +309,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
         }
         LOG.debug("clientID {} is a newcome, creating it's empty subscriptions set", clientID);
         m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
-        return new ClientSession(clientID, this);
+        return new ClientSession(clientID, this, this);
     }
 
     @Override
@@ -318,7 +318,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore {
             return null;
         }
 
-        return new ClientSession(clientID, this);
+        return new ClientSession(clientID, this, this);
     }
 
     @Override