From 635c87389d7178d72a539afdc753225f93e15ebd Mon Sep 17 00:00:00 2001
From: andsel <selva.andre@gmail.com>
Date: Sat, 31 Oct 2015 11:34:15 +0100
Subject: [PATCH] Refactored ProtocolProcessor.processUnsubscribe to use the
 ClientSession concept instead of direct access to map storage

---
 .../java/org/eclipse/moquette/spi/ClientSession.java | 12 ++++++++++++
 .../eclipse/moquette/spi/impl/ProtocolProcessor.java |  3 +++
 .../spi/impl/subscriptions/SubscriptionsStore.java   |  1 -
 3 files changed, 15 insertions(+), 1 deletion(-)

diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java
index 98d60187..e07c33d2 100644
--- a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java
+++ b/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);
+    }
 }
\ No newline at end of file
diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
index fca6e847..c3b41c80 100644
--- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java
+++ b/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);
         }
 
diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java
index 3e80b622..17b3fff6 100644
--- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java
+++ b/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);
     }
     
     /**