Refactory, moved SubscriptionStore related method directly in the subscriptioStore from ProtocolProcessro

ChangeLog.txt

@@ -10,6 +10,8 @@ Version 0.8:
     Changed willMessage and password in CONNECT message from String to byte[] (fix issue #94 and #95)
     Fixed ugly empty session storage, avoid fake NullObject empty subscription subscription
     Added check to avoid creation of LostConnectionEvents for not yet completed connects (fix issue #88)
+    Removed LMAX ringbuffer to use multi threading with persistent (copy on write) data structures
+    Fixed bug that on connection stolen dropped also the new connection (fix issue #108)
 
 Version 0.7:
     Fix issue #38, added aligned to MQTT 3.1.1

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

@@ -274,7 +274,6 @@ public class ProtocolProcessor {
     private void cleanSession(String clientID) {
         LOG.info("cleaning old saved subscriptions for client <{}>", clientID);
         //remove from log all subscriptions
-        //m_sessionsStore.wipeSubscriptions(clientID);
         subscriptions.removeForClient(clientID);
 
         //remove also the messages stored of type QoS1/2
@@ -578,7 +577,6 @@ public class ProtocolProcessor {
             }
 
             subscriptions.removeSubscription(topic, clientID);
-            m_sessionsStore.removeSubscription(topic, clientID);
             m_interceptor.notifyTopicUnsubscribed(topic, clientID);
         }
 
@@ -619,7 +617,6 @@ public class ProtocolProcessor {
             //send SUBACK with 0x80 for this topic filter
             return false;
         }
-        m_sessionsStore.addNewSubscription(newSubscription);
         subscriptions.add(newSubscription);
 
         //notify the Observables

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

@@ -179,6 +179,7 @@ public class SubscriptionsStore {
     }
 
     public void add(Subscription newSubscription) {
+        m_sessionsStore.addNewSubscription(newSubscription);
         addDirect(newSubscription);
     }
 
@@ -205,6 +206,7 @@ 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);
     }
     
     /**
@@ -213,7 +215,7 @@ public class SubscriptionsStore {
     public void clearAllSubscriptions() {
         SubscriptionTreeCollector subsCollector = new SubscriptionTreeCollector();
         bfsVisit(subscriptions.get(), subsCollector, 0);
-        
+
         List<Subscription> allSubscriptions = subsCollector.getResult();
         for (Subscription subscription : allSubscriptions) {
             removeSubscription(subscription.getTopicFilter(), subscription.getClientId());

broker/src/test/java/org/eclipse/moquette/spi/impl/MemorySessionStore.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright (c) 2012-2015 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ * The Eclipse Public License is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * The Apache License v2.0 is available at
+ * http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package org.eclipse.moquette.spi.impl;
+
+import org.eclipse.moquette.spi.ISessionsStore;
+import org.eclipse.moquette.spi.impl.subscriptions.Subscription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+
+/**
+ *
+ * @author andrea
+ */
+public class MemorySessionStore implements ISessionsStore {
+    private static final Logger LOG = LoggerFactory.getLogger(MemorySessionStore.class);
+
+    private Map<String, Set<Subscription>> m_persistentSubscriptions = new HashMap<>();
+
+    @Override
+    public void removeSubscription(String topic, String clientID) {
+        LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID);
+        if (!m_persistentSubscriptions.containsKey(clientID)) {
+            return;
+        }
+        Set<Subscription> clientSubscriptions = m_persistentSubscriptions.get(clientID);
+        //search for the subscription to remove
+        Subscription toBeRemoved = null;
+        for (Subscription sub : clientSubscriptions) {
+            if (sub.getTopicFilter().equals(topic)) {
+                toBeRemoved = sub;
+                break;
+            }
+        }
+
+        if (toBeRemoved != null) {
+            clientSubscriptions.remove(toBeRemoved);
+        }
+    }
+
+    @Override
+    public void addNewSubscription(Subscription newSubscription) {
+        final String clientID = newSubscription.getClientId();
+        if (!m_persistentSubscriptions.containsKey(clientID)) {
+            m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
+        }
+
+        Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
+        if (!subs.contains(newSubscription)) {
+            subs.add(newSubscription);
+            m_persistentSubscriptions.put(clientID, subs);
+        }
+    }
+
+    @Override
+    public void wipeSubscriptions(String clientID) {
+        m_persistentSubscriptions.remove(clientID);
+    }
+
+    @Override
+    public void updateSubscriptions(String clientID, Set<Subscription> subscriptions) {
+        m_persistentSubscriptions.put(clientID, subscriptions);
+    }
+
+    @Override
+    public boolean contains(String clientID) {
+        return m_persistentSubscriptions.containsKey(clientID);
+    }
+
+    @Override
+    public void createNewSession(String clientID) {
+        if (m_persistentSubscriptions.containsKey(clientID)) {
+            LOG.error("already exists a session for client <{}>", clientID);
+            return;
+        }
+        m_persistentSubscriptions.put(clientID, new HashSet<Subscription>());
+    }
+
+    @Override
+    public List<Subscription> listAllSubscriptions() {
+        List<Subscription> allSubscriptions = new ArrayList<Subscription>();
+        for (Map.Entry<String, Set<Subscription>> entry : m_persistentSubscriptions.entrySet()) {
+            allSubscriptions.addAll(entry.getValue());
+        }
+        return allSubscriptions;
+    }
+}

broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java

@@ -459,6 +459,7 @@ public class ProtocolProcessorTest {
         Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE, false);
         Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE, false);
         SubscriptionsStore subscriptions = new SubscriptionsStore();
+        subscriptions.init(new MemorySessionStore());
         subscriptions.add(subQos1);
         subscriptions.add(subQos2);