Refactored ProtocolProcessor.processUnsubscribe to use the ClientSession concept instead of direct access to map storage

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

@@ -22,6 +22,7 @@ import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -84,4 +85,15 @@ public class ClientSession {
         m_sessionsStore.addNewSubscription(newSubscription);
         return true;
     }
+
+    public void unsubscribeFrom(String topicFilter) {
+        m_sessionsStore.removeSubscription(topicFilter, clientID);
+        Set<Subscription> subscriptionsToRemove = new HashSet<>();
+        for (Subscription sub : this.subscriptions) {
+            if (sub.getTopicFilter().equals(topicFilter)) {
+                subscriptionsToRemove.add(sub);
+            }
+        }
+        subscriptions.removeAll(subscriptionsToRemove);
+    }
 }

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

@@ -568,6 +568,8 @@ public class ProtocolProcessor {
         String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
         LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID);
 
+        ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
+
         for (String topic : topics) {
             boolean validTopic = SubscriptionsStore.validate(topic);
             if (!validTopic) {
@@ -578,6 +580,7 @@ public class ProtocolProcessor {
             }
 
             subscriptions.removeSubscription(topic, clientID);
+            clientSession.unsubscribeFrom(topic);
             m_interceptor.notifyTopicUnsubscribed(topic, clientID);
         }
 

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

@@ -198,7 +198,6 @@ public class SubscriptionsStore {
             }
             //spin lock repeating till we can, swap root, if can't swap just re-do the operation
         } while(!subscriptions.compareAndSet(oldRoot, couple.root));
-        m_sessionsStore.removeSubscription(topic, clientID);
     }
     
     /**