Changed the subscription tree to work as persistent tree

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

@@ -20,6 +20,7 @@ import com.lmax.disruptor.RingBuffer;
 import com.lmax.disruptor.dsl.Disruptor;
 import java.nio.ByteBuffer;
 import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -96,7 +97,7 @@ public class ProtocolProcessor implements EventHandler<ValueEvent> {
     
     private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
     
-    private Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<>();
+    private Map<String, ConnectionDescriptor> m_clientIDs = new ConcurrentHashMap<>();
     private SubscriptionsStore subscriptions;
     private boolean allowAnonymous;
     private IAuthorizator m_authorizator;
@@ -106,7 +107,7 @@ public class ProtocolProcessor implements EventHandler<ValueEvent> {
     private BrokerInterceptor m_interceptor;
 
     //maps clientID to Will testament, if specified on CONNECT
-    private Map<String, WillMessage> m_willStore = new HashMap<>();
+    private Map<String, WillMessage> m_willStore = new ConcurrentHashMap<>();
     
     private ExecutorService m_executor;
     private RingBuffer<ValueEvent> m_ringBuffer;

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

@@ -18,6 +18,7 @@ package org.eclipse.moquette.spi.impl.subscriptions;
 import java.text.ParseException;
 import java.util.*;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.eclipse.moquette.spi.ISessionsStore;
 import org.slf4j.Logger;
@@ -30,6 +31,16 @@ import org.slf4j.LoggerFactory;
  */
 public class SubscriptionsStore {
 
+    public static class NodeCouple {
+        final TreeNode root;
+        final TreeNode createdNode;
+
+        public NodeCouple(TreeNode root, TreeNode createdNode) {
+            this.root = root;
+            this.createdNode = createdNode;
+        }
+    }
+
     /**
      * Check if the topic filter of the subscription is well formed
      * */
@@ -43,7 +54,7 @@ public class SubscriptionsStore {
         }
     }
 
-    public static interface IVisitor<T> {
+    public interface IVisitor<T> {
         void visit(TreeNode node, int deep);
         
         T getResult();
@@ -90,7 +101,7 @@ public class SubscriptionsStore {
         }
     }
 
-    private TreeNode subscriptions = new TreeNode(null);
+    private AtomicReference<TreeNode> subscriptions = new AtomicReference<>(new TreeNode(null));
     private ISessionsStore m_sessionsStore;
     private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
 
@@ -115,12 +126,19 @@ public class SubscriptionsStore {
             LOG.debug("Finished loading. Subscription tree after {}", dumpTree());
         }
     }
-    
+
+    //TODO modify the data structure
     protected void addDirect(Subscription newSubscription) {
-        TreeNode current = findMatchingNode(newSubscription.topicFilter);
-        current.addSubscription(newSubscription);
+        TreeNode oldRoot;
+        NodeCouple couple;
+        do {
+            oldRoot = subscriptions.get();
+            couple = recreatePath(newSubscription.topicFilter, oldRoot);
+            couple.createdNode.addSubscription(newSubscription); //createdNode could be null?
+            //spin lock repeating till we can, swap root, if can't swap just re-do the operation
+        } while(!subscriptions.compareAndSet(oldRoot, couple.root));
     }
-    
+
     private TreeNode findMatchingNode(String topic) {
         List<Token> tokens = new ArrayList<>();
         try {
@@ -131,13 +149,44 @@ public class SubscriptionsStore {
 //            return;
         }
 
-        TreeNode current = subscriptions;
+        TreeNode current = subscriptions.get();
         for (Token token : tokens) {
             TreeNode matchingChildren;
 
             //check if a children with the same token already exists
             if ((matchingChildren = current.childWithToken(token)) != null) {
                 current = matchingChildren;
+            } else {
+                //can't find any match
+                return null;
+            }
+        }
+        return current;
+    }
+
+    protected NodeCouple recreatePath(String topic, final TreeNode oldRoot) {
+        List<Token> tokens = new ArrayList<>();
+        try {
+            tokens = parseTopic(topic);
+        } catch (ParseException ex) {
+            //TODO handle the parse exception
+            LOG.error(null, ex);
+        }
+
+        final TreeNode newRoot = oldRoot.copy();
+        TreeNode parent = newRoot;
+        TreeNode current = newRoot;
+        for (Token token : tokens) {
+            TreeNode matchingChildren;
+
+            //check if a children with the same token already exists
+            if ((matchingChildren = current.childWithToken(token)) != null) {
+                //copy the traversed node
+                current = matchingChildren.copy();
+                current.m_parent = parent;
+                //update the child just added in the children list
+                parent.updateChild(matchingChildren, current);
+                parent = current;
             } else {
                 //create a new node for the newly inserted token
                 matchingChildren = new TreeNode(current);
@@ -146,15 +195,11 @@ public class SubscriptionsStore {
                 current = matchingChildren;
             }
         }
-        return current;
+        return new NodeCouple(newRoot, current);
     }
 
     public void add(Subscription newSubscription) {
         addDirect(newSubscription);
-
-        //log the subscription
-//        String clientID = newSubscription.getClientId();
-//        m_storageService.addNewSubscription(newSubscription, clientID);
     }
 
 
@@ -176,15 +221,15 @@ public class SubscriptionsStore {
     }
     
     /**
-     * TODO implement testing
      * It's public because needed in tests :-( bleah
      */
     public void clearAllSubscriptions() {
         SubscriptionTreeCollector subsCollector = new SubscriptionTreeCollector();
-        bfsVisit(subscriptions, subsCollector, 0);
+        bfsVisit(subscriptions.get(), subsCollector, 0);
         
         List<Subscription> allSubscriptions = subsCollector.getResult();
         for (Subscription subscription : allSubscriptions) {
+            //TODO modify the data structure
             removeSubscription(subscription.getTopicFilter(), subscription.getClientId());
         }
     }
@@ -192,24 +237,27 @@ public class SubscriptionsStore {
     /**
      * Visit the topics tree to remove matching subscriptions with clientID
      */
+    //TODO modify the data structure
     public void removeForClient(String clientID) {
-        subscriptions.removeClientSubscriptions(clientID);
+        subscriptions.get().removeClientSubscriptions(clientID);
         //persist the update
         m_sessionsStore.wipeSubscriptions(clientID);
     }
 
+    //TODO modify the data structure
     public void deactivate(String clientID) {
-        subscriptions.deactivate(clientID);
+        subscriptions.get().deactivate(clientID);
         //persist the update
-        Set<Subscription> subs = subscriptions.findAllByClientID(clientID);
+        Set<Subscription> subs = subscriptions.get().findAllByClientID(clientID);
         m_sessionsStore.updateSubscriptions(clientID, subs);
     }
 
+    //TODO modify the data structure
     public void activate(String clientID) {
         LOG.debug("Activating subscriptions for clientID <{}>", clientID);
-        subscriptions.activate(clientID);
+        subscriptions.get().activate(clientID);
         //persist the update
-        Set<Subscription> subs = subscriptions.findAllByClientID(clientID);
+        Set<Subscription> subs = subscriptions.get().findAllByClientID(clientID);
         m_sessionsStore.updateSubscriptions(clientID, subs);
     }
 
@@ -230,7 +278,7 @@ public class SubscriptionsStore {
 
         Queue<Token> tokenQueue = new LinkedBlockingDeque<>(tokens);
         List<Subscription> matchingSubs = new ArrayList<>();
-        subscriptions.matches(tokenQueue, matchingSubs);
+        subscriptions.get().matches(tokenQueue, matchingSubs);
 
         //remove the overlapping subscriptions, selecting ones with greatest qos
         Map<String, Subscription> subsForClient = new HashMap<>();
@@ -249,12 +297,12 @@ public class SubscriptionsStore {
     }
 
     public int size() {
-        return subscriptions.size();
+        return subscriptions.get().size();
     }
     
     public String dumpTree() {
         DumpTreeVisitor visitor = new DumpTreeVisitor();
-        bfsVisit(subscriptions, visitor, 0);
+        bfsVisit(subscriptions.get(), visitor, 0);
         return visitor.getResult();
     }
     

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

@@ -23,7 +23,7 @@ public class Token {
     static final Token EMPTY = new Token("");
     static final Token MULTI = new Token("#");
     static final Token SINGLE = new Token("+");
-    String name;
+    final String name;
 
     protected Token(String s) {
         name = s;

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

@@ -64,9 +64,22 @@ class TreeNode {
     void addChild(TreeNode child) {
         m_children.add(child);
     }
+//
+//    boolean isLeaf() {
+//        return m_children.isEmpty();
+//    }
 
-    boolean isLeaf() {
-        return m_children.isEmpty();
+    /**
+     * Creates a shallow copy of the current node.
+     * Copy the token and the children.
+     * */
+    TreeNode copy() {
+        final TreeNode copy = new TreeNode(this);
+        copy.m_parent = m_parent;
+        copy.m_children = new ArrayList<>(m_children);
+        copy.m_subscriptions = new ArrayList<>(m_subscriptions);
+        copy.m_token = m_token;
+        return copy;
     }
 
     /**
@@ -83,6 +96,11 @@ class TreeNode {
         return null;
     }
 
+    void updateChild(TreeNode oldChild, TreeNode newChild) {
+        m_children.remove(oldChild);
+        m_children.add(newChild);
+    }
+
     List<Subscription> subscriptions() {
         return m_subscriptions;
     }

broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java

@@ -76,7 +76,7 @@ public class SubscriptionsStoreTest {
     }
 
     @Test(expected = ParseException.class)
-    public void testParseTopicMultiNotAferSeparatorNotValid() throws ParseException {
+    public void testParseTopicMultiNotAfterSeparatorNotValid() throws ParseException {
         store.parseTopic("finance#");
     }
 
@@ -397,6 +397,32 @@ public class SubscriptionsStoreTest {
         assertFalse(subscriptions.contains(client1SubQoS0)); //client1SubQoS2 should override client1SubQoS0
     }
 
+    @Test
+    public void testRecreatePath_emptyRoot() {
+        TreeNode oldRoot = new TreeNode(null);
+        final SubscriptionsStore.NodeCouple resp = store.recreatePath("/finance", oldRoot);
+
+        //Verify
+        assertNotNull(resp.root);
+        assertNull(resp.root.m_token);
+        assertEquals(1, resp.root.m_children.size());
+        assertEquals(resp.createdNode, resp.root.m_children.get(0).m_children.get(0));
+    }
+
+    @Test
+    public void testRecreatePath_1layer_tree() {
+        TreeNode oldRoot = new TreeNode(null);
+        final SubscriptionsStore.NodeCouple respFinance = store.recreatePath("/finance", oldRoot);
+        final SubscriptionsStore.NodeCouple respPlus = store.recreatePath("/+", respFinance.root);
+
+        //Verify
+        assertNotNull(respPlus.root);
+        assertNull(respPlus.root.m_token);
+        assertEquals(1, respPlus.root.m_children.size());
+        assertTrue(respPlus.root.m_children.get(0).m_children.contains(respPlus.createdNode));
+        assertTrue(respPlus.root.m_children.get(0).m_children.contains(respFinance.createdNode));
+    }
+
     private static Token[] asArray(Object... l) {
         Token[] tokens = new Token[l.length];
         for (int i = 0; i < l.length; i++) {