Extracted common subscriptions directory interface and switched to CTrie inspired implementation

broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java

@@ -20,8 +20,8 @@ import io.moquette.server.ConnectionDescriptorStore;
 import io.moquette.spi.ClientSession;
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.ISessionsStore;
+import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
 import io.moquette.spi.impl.subscriptions.Subscription;
-import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
 import io.moquette.spi.impl.subscriptions.Topic;
 import io.netty.buffer.ByteBuf;
 import io.netty.handler.codec.mqtt.*;
@@ -36,10 +36,10 @@ class MessagesPublisher {
     private final ConnectionDescriptorStore connectionDescriptors;
     private final ISessionsStore m_sessionsStore;
     private final PersistentQueueMessageSender messageSender;
-    private final SubscriptionsDirectory subscriptions;
+    private final ISubscriptionsDirectory subscriptions;
 
     public MessagesPublisher(ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
-                             PersistentQueueMessageSender messageSender, SubscriptionsDirectory subscriptions) {
+                             PersistentQueueMessageSender messageSender, ISubscriptionsDirectory subscriptions) {
         this.connectionDescriptors = connectionDescriptors;
         this.m_sessionsStore = sessionsStore;
         this.messageSender = messageSender;

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java

@@ -23,8 +23,8 @@ import io.moquette.server.ConnectionDescriptorStore;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.*;
 import io.moquette.spi.IMessagesStore.StoredMessage;
+import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
 import io.moquette.spi.impl.subscriptions.Subscription;
-import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
 import io.moquette.spi.impl.subscriptions.Topic;
 import io.moquette.spi.security.IAuthenticator;
 import io.moquette.spi.security.IAuthorizator;
@@ -128,7 +128,7 @@ public class ProtocolProcessor {
     protected ConnectionDescriptorStore connectionDescriptors;
     protected ConcurrentMap<RunningSubscription, SubscriptionState> subscriptionInCourse;
 
-    private SubscriptionsDirectory subscriptions;
+    private ISubscriptionsDirectory subscriptions;
     private ISubscriptionsStore subscriptionStore;
     private boolean allowAnonymous;
     private boolean allowZeroByteClientId;
@@ -153,21 +153,21 @@ public class ProtocolProcessor {
     ProtocolProcessor() {
     }
 
-    public void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
+    public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                      IAuthenticator authenticator, boolean allowAnonymous, IAuthorizator authorizator,
                      BrokerInterceptor interceptor) {
         init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, false, authorizator,
             interceptor, null);
     }
 
-    public void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
+    public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                      IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
                      IAuthorizator authorizator, BrokerInterceptor interceptor) {
         init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, allowZeroByteClientId,
             authorizator, interceptor, null);
     }
 
-    public void init(SubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
+    public void init(ISubscriptionsDirectory subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore,
                      IAuthenticator authenticator, boolean allowAnonymous, boolean allowZeroByteClientId,
                      IAuthorizator authorizator, BrokerInterceptor interceptor, String serverPort) {
         init(new ConnectionDescriptorStore(sessionsStore), subscriptions, storageService, sessionsStore, authenticator,
@@ -192,7 +192,7 @@ public class ProtocolProcessor {
      * @param interceptor
      *            to notify events to an intercept handler
      */
-    void init(ConnectionDescriptorStore connectionDescriptors, SubscriptionsDirectory subscriptions,
+    void init(ConnectionDescriptorStore connectionDescriptors, ISubscriptionsDirectory subscriptions,
             IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator,
             boolean allowAnonymous, boolean allowZeroByteClientId, IAuthorizator authorizator,
             BrokerInterceptor interceptor, String serverPort) {

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java

@@ -27,6 +27,8 @@ import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.IStore;
 import io.moquette.spi.ISubscriptionsStore;
 import io.moquette.spi.impl.security.*;
+import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory;
+import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
 import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
 import io.moquette.spi.security.IAuthenticator;
@@ -118,7 +120,7 @@ public class ProtocolProcessorBootstrapper {
         BrokerInterceptor interceptor = new BrokerInterceptor(props, observers);
 
         LOG.info("Initializing subscriptions store...");
-        SubscriptionsDirectory subscriptions = new SubscriptionsDirectory();
+        ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
         subscriptions.init(m_sessionsStore);
 
         LOG.info("Configuring MQTT authenticator...");

broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java

@@ -21,7 +21,7 @@ import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.ClientSession;
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.ISessionsStore;
-import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
+import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
 import io.moquette.spi.impl.subscriptions.Topic;
 import io.moquette.spi.security.IAuthorizator;
 import io.netty.channel.Channel;
@@ -39,14 +39,14 @@ class Qos2PublishHandler extends QosPublishHandler {
 
     private static final Logger LOG = LoggerFactory.getLogger(Qos2PublishHandler.class);
 
-    private final SubscriptionsDirectory subscriptions;
+    private final ISubscriptionsDirectory subscriptions;
     private final IMessagesStore m_messagesStore;
     private final BrokerInterceptor m_interceptor;
     private final ConnectionDescriptorStore connectionDescriptors;
     private final ISessionsStore m_sessionsStore;
     private final MessagesPublisher publisher;
 
-    public Qos2PublishHandler(IAuthorizator authorizator, SubscriptionsDirectory subscriptions,
+    public Qos2PublishHandler(IAuthorizator authorizator, ISubscriptionsDirectory subscriptions,
                               IMessagesStore messagesStore, BrokerInterceptor interceptor,
                               ConnectionDescriptorStore connectionDescriptors, ISessionsStore sessionsStore,
                               MessagesPublisher messagesPublisher) {

broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright (c) 2012-2017 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 io.moquette.spi.impl.subscriptions;
+
+import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
+
+import java.util.*;
+
+class CNode {
+
+    Token token;
+    private List<INode> children;
+    Set<ClientTopicCouple> subscriptions = new HashSet<>();
+
+    //private int subtreeSubscriptions;
+
+    CNode() {
+        this.children = new ArrayList<>();
+        this.subscriptions = new HashSet<>();
+    }
+
+    //Copy constructor
+    private CNode(Token token, List<INode> children, Set<ClientTopicCouple> subscriptions) {
+        this.token = token;
+        this.children = children;
+        this.subscriptions = subscriptions;
+    }
+
+    boolean anyChildrenMatch(Token token) {
+        for (INode iNode : children) {
+            final CNode child = iNode.mainNode();
+            if (child.equals(token)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    List<INode> allChildren() {
+        return this.children;
+    }
+
+    INode childOf(Token token) {
+        for (INode iNode : children) {
+            final CNode child = iNode.mainNode();
+            if (child.equals(token)) {
+                return iNode;
+            }
+        }
+        throw new IllegalArgumentException("Asked for a token that doesn't exists in any child [" + token + "]");
+    }
+
+    private boolean equals(Token token) {
+        if (this.token == null) {
+            return false;
+        }
+        return this.token.equals(token);
+    }
+
+    CNode copy() {
+        return new CNode(this.token, this.children, this.subscriptions);
+    }
+
+    public void add(INode newINode) {
+        this.children.add(newINode);
+    }
+
+    CNode addSubscription(String clientId, Topic topic) {
+        this.subscriptions.add(new ClientTopicCouple(clientId, topic));
+        return this;
+    }
+
+    /**
+     * @return true iff the subscriptions contained in this node are owned by clientId
+     * */
+    boolean containsOnly(String clientId) {
+        for (ClientTopicCouple sub : this.subscriptions) {
+            if (!sub.clientID.equals(clientId)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    //TODO this is equivalent to negate(containsOnly(clientId))
+    public boolean contains(String clientId) {
+        for (ClientTopicCouple sub : this.subscriptions) {
+            if (sub.clientID.equals(clientId)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    void removeSubscriptionsFor(String clientId) {
+        Set<ClientTopicCouple> toRemove = new HashSet<>();
+        for (ClientTopicCouple sub : this.subscriptions) {
+            if (sub.clientID.equals(clientId)) {
+                toRemove.add(sub);
+            }
+        }
+        this.subscriptions.removeAll(toRemove);
+    }
+
+//    public List<INode> childrenMatching(Token token) {
+//        List<INode> res = new ArrayList<>();
+//        for (INode iNode : children) {
+//            final CNode child = iNode.mainNode();
+//            if (child.equals(token) || (child.token == Token.MULTI || child.token == Token.SINGLE)) {
+//                res.add(iNode);
+//            }
+//        }
+//        return res;
+//    }
+}

broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java

@@ -0,0 +1,322 @@
+/*
+ * Copyright (c) 2012-2017 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 io.moquette.spi.impl.subscriptions;
+
+import io.moquette.spi.ISessionsStore;
+import io.moquette.spi.ISubscriptionsStore;
+import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
+import io.netty.util.internal.StringUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory {
+
+    private static final Logger LOG = LoggerFactory.getLogger(CTrieSubscriptionDirectory.class);
+
+    private static final Token ROOT = new Token("root");
+
+    protected INode root;
+    private volatile ISubscriptionsStore subscriptionsStore;
+
+    private interface IVisitor<T> {
+
+        void visit(CNode node, int deep);
+
+        T getResult();
+    }
+
+    private class DumpTreeVisitor implements IVisitor<String> {
+
+        String s = "";
+
+        @Override
+        public void visit(CNode node, int deep) {
+            String indentTabs = indentTabs(deep);
+            s += indentTabs + (node.token == null ? "" : node.token.toString()) + prettySubscriptions(node) + "\n";
+        }
+
+        private String prettySubscriptions(CNode node) {
+            if (node.subscriptions.isEmpty()) {
+                return StringUtil.EMPTY_STRING;
+            }
+            StringBuilder subScriptionsStr = new StringBuilder(" ~~[");
+            int counter = 0;
+            for (ClientTopicCouple couple : node.subscriptions) {
+                subScriptionsStr
+                    .append("{filter=").append(couple.topicFilter).append(", ")
+                    .append("client='").append(couple.clientID).append("'}");
+                counter++;
+                if (counter < node.subscriptions.size()) {
+                    subScriptionsStr.append(";");
+                }
+            }
+            return subScriptionsStr.append("]").toString();
+        }
+
+        private String indentTabs(int deep) {
+            StringBuilder s = new StringBuilder();
+            if (deep > 0) {
+                for (int i = 0; i < deep - 1; i++) {
+                    s.append("| ");
+                }
+                s.append("|-");
+            }
+            return s.toString();
+        }
+
+        @Override
+        public String getResult() {
+            return s;
+        }
+    }
+
+    private enum Action {
+        OK, REPEAT
+    }
+
+    public void init(ISessionsStore sessionsStore) {
+        LOG.info("Initializing CTrie");
+        final CNode mainNode = new CNode();
+        mainNode.token = ROOT;
+        this.root = new INode(mainNode);
+
+        LOG.info("Initializing subscriptions store...");
+        this.subscriptionsStore = sessionsStore.subscriptionStore();
+        List<ClientTopicCouple> subscriptions = this.subscriptionsStore.listAllSubscriptions();
+        // reload any subscriptions persisted
+        if (LOG.isTraceEnabled()) {
+            LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}", dumpTree());
+        }
+
+        for (ClientTopicCouple clientTopic : subscriptions) {
+            LOG.info("Re-subscribing client to topic CId={}, topicFilter={}", clientTopic.clientID,
+                clientTopic.topicFilter);
+            add(clientTopic);
+        }
+        if (LOG.isTraceEnabled()) {
+            LOG.trace("Stored subscriptions have been reloaded. SubscriptionTree = {}", dumpTree());
+        }
+    }
+
+    Optional<CNode> lookup(Topic topic) {
+        INode inode = this.root;
+        Token token = topic.headToken();
+        while (!topic.isEmpty() && (inode.mainNode().anyChildrenMatch(token))) {
+            topic = topic.exceptHeadToken();
+            inode = inode.mainNode().childOf(token);
+            token = topic.headToken();
+        }
+        if (inode == null || !topic.isEmpty()) {
+            return Optional.empty();
+        }
+        return Optional.of(inode.mainNode());
+    }
+
+
+    @Override
+    public List<Subscription> matches(Topic topic) {
+        return new ArrayList<>(match(topic));
+    }
+
+
+    /**
+     * Given a topic string return the clients subscriptions that matches it. Topic string can't
+     * contain character # and + because they are reserved to listeners subscriptions, and not topic
+     * publishing.
+     *
+     * @param topic
+     *            to use fo searching matching subscriptions.
+     * @return the list of matching subscriptions, or empty if not matching.
+     */
+     Set<Subscription> match(Topic topic) {
+        final Set<ClientTopicCouple> matchingSubs = recursiveMatch(topic, this.root);
+        // remove the overlapping subscriptions, selecting ones with greatest qos
+        Map<String, Subscription> subsForClient = new HashMap<>();
+        for (ClientTopicCouple matchingCouple : matchingSubs) {
+            Subscription existingSub = subsForClient.get(matchingCouple.clientID);
+            Subscription sub = this.subscriptionsStore.getSubscription(matchingCouple);
+            if (sub == null) {
+                // if the m_sessionStore hasn't the sub because the client disconnected
+                continue;
+            }
+            // update the selected subscriptions if not present or if has a greater qos
+            if (existingSub == null || existingSub.getRequestedQos().value() < sub.getRequestedQos().value()) {
+                subsForClient.put(matchingCouple.clientID, sub);
+            }
+        }
+        return new HashSet<>(subsForClient.values());
+    }
+
+    Set<ClientTopicCouple> recursiveMatch(Topic topic, INode inode) {
+        CNode cnode = inode.mainNode();
+        if (cnode.token == Token.MULTI) {
+            return cnode.subscriptions;
+        }
+        if (topic.isEmpty()) {
+            return Collections.emptySet();
+        }
+        if (cnode instanceof TNode) {
+            return Collections.emptySet();
+        }
+        final Token token = topic.headToken();
+        if (!(cnode.token == Token.SINGLE || cnode.token.equals(token) || cnode.token == ROOT)) {
+            return Collections.emptySet();
+        }
+        Topic remainingTopic = (cnode.token == ROOT) ? topic : topic.exceptHeadToken();
+        Set<ClientTopicCouple> subscriptions = new HashSet<>();
+        if (remainingTopic.isEmpty()) {
+            subscriptions.addAll(cnode.subscriptions);
+        }
+        for (INode subInode : cnode.allChildren()) {
+            subscriptions.addAll(recursiveMatch(remainingTopic, subInode));
+        }
+        return subscriptions;
+    }
+
+
+    public void add(ClientTopicCouple newSubscription) {
+        Action res;
+        do {
+            res = insert(newSubscription.clientID, newSubscription.topicFilter, this.root, newSubscription.topicFilter);
+        } while (res == Action.REPEAT);
+    }
+
+    private Action insert(String clientId, Topic topic, final INode inode, Topic fullpath) {
+        Token token = topic.headToken();
+        if (!topic.isEmpty() && inode.mainNode().anyChildrenMatch(token)) {
+            Topic remainingTopic = topic.exceptHeadToken();
+            INode nextInode = inode.mainNode().childOf(token);
+            return insert(clientId, remainingTopic, nextInode, fullpath);
+        } else {
+            if (topic.isEmpty()) {
+                return insertSubscription(clientId, fullpath, inode);
+            } else {
+                return createNodeAndInsertSubscription(clientId, topic, inode, fullpath);
+            }
+        }
+    }
+
+    private Action insertSubscription(String clientId, Topic topic, INode inode) {
+        CNode cnode = inode.mainNode();
+        CNode updatedCnode = cnode.copy().addSubscription(clientId, topic);
+        if (inode.compareAndSet(cnode, updatedCnode)) {
+            return Action.OK;
+        } else {
+            return Action.REPEAT;
+        }
+    }
+
+    private Action createNodeAndInsertSubscription(String clientId, Topic topic, INode inode, Topic fullpath) {
+        INode newInode = createPathRec(clientId, topic, fullpath);
+        CNode cnode = inode.mainNode();
+        CNode updatedCnode = cnode.copy();
+        updatedCnode.add(newInode);
+
+        return inode.compareAndSet(cnode, updatedCnode) ? Action.OK : Action.REPEAT;
+    }
+
+    private INode createLeafNodes(String clientId, Topic fullpath, Token token) {
+        CNode newLeafCnode = new CNode();
+        newLeafCnode.token = token;
+        newLeafCnode.addSubscription(clientId, fullpath);
+
+        return new INode(newLeafCnode);
+    }
+
+    private INode createPathRec(String clientId, Topic topic, Topic fullpath) {
+        Topic remainingTopic = topic.exceptHeadToken();
+        if (!remainingTopic.isEmpty()) {
+            INode inode = createPathRec(clientId, remainingTopic, fullpath);
+            CNode cnode = new CNode();
+            cnode.token = topic.headToken();
+            cnode.add(inode);
+            return new INode(cnode);
+        } else {
+            return createLeafNodes(clientId, fullpath, topic.headToken());
+        }
+    }
+
+    public void removeSubscription(Topic topic, String clientID) {
+        Action res;
+        do {
+            res = remove(clientID, topic, this.root);
+        } while (res == Action.REPEAT);
+    }
+
+    private Action remove(String clientId, Topic topic, INode inode) {
+        Token token = topic.headToken();
+        if (!topic.isEmpty() && (inode.mainNode().anyChildrenMatch(token))) {
+            Topic remainingTopic = topic.exceptHeadToken();
+            INode nextInode = inode.mainNode().childOf(token);
+            return remove(clientId, remainingTopic, nextInode);
+        } else {
+            final CNode cnode = inode.mainNode();
+            if (cnode.containsOnly(clientId)) {
+                TNode tnode = new TNode();
+                return inode.compareAndSet(cnode, tnode) ? Action.OK : Action.REPEAT;
+            } else if (cnode.contains(clientId)) {
+                CNode updatedCnode = cnode.copy();
+                updatedCnode.removeSubscriptionsFor(clientId);
+                return inode.compareAndSet(cnode, updatedCnode) ? Action.OK : Action.REPEAT;
+            } else {
+                //someone else already removed
+                return Action.OK;
+            }
+        }
+    }
+
+    private class SubscriptionCounterVisitor implements IVisitor<Integer> {
+
+        AtomicInteger accumulator = new AtomicInteger(0);
+
+        @Override
+        public void visit(CNode node, int deep) {
+            accumulator.addAndGet(node.subscriptions.size());
+        }
+
+        @Override
+        public Integer getResult() {
+            return accumulator.get();
+        }
+    }
+
+    public int size() {
+        SubscriptionCounterVisitor visitor = new SubscriptionCounterVisitor();
+        dfsVisit(this.root, visitor, 0);
+        return visitor.getResult();
+    }
+
+    public String dumpTree() {
+        DumpTreeVisitor visitor = new DumpTreeVisitor();
+        dfsVisit(this.root, visitor, 0);
+        return visitor.getResult();
+    }
+
+    private void dfsVisit(INode node, IVisitor<?> visitor, int deep) {
+        if (node == null) {
+            return;
+        }
+
+        visitor.visit(node.mainNode(), deep);
+        ++deep;
+        for (INode child : node.mainNode().allChildren()) {
+            dfsVisit(child, visitor, deep);
+        }
+    }
+}

broker/src/main/java/io/moquette/spi/impl/subscriptions/INode.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright (c) 2012-2017 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 io.moquette.spi.impl.subscriptions;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+class INode {
+    private AtomicReference<CNode> mainNode = new AtomicReference<>();
+
+    INode(CNode mainNode) {
+        this.mainNode.set(mainNode);
+    }
+
+    boolean compareAndSet(CNode old, CNode newNode) {
+        return mainNode.compareAndSet(old, newNode);
+    }
+
+    boolean compareAndSet(CNode old, TNode newNode) {
+        return mainNode.compareAndSet(old, newNode);
+    }
+
+    CNode mainNode() {
+        return this.mainNode.get();
+    }
+}

broker/src/main/java/io/moquette/spi/impl/subscriptions/ISubscriptionsDirectory.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright (c) 2012-2017 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 io.moquette.spi.impl.subscriptions;
+
+import io.moquette.spi.ISessionsStore;
+import io.moquette.spi.ISubscriptionsStore;
+
+import java.util.List;
+
+public interface ISubscriptionsDirectory {
+
+    void init(ISessionsStore sessionsStore);
+
+    void add(ISubscriptionsStore.ClientTopicCouple newSubscription);
+
+    void removeSubscription(Topic topic, String clientID);
+
+    List<Subscription> matches(Topic topic);
+
+    int size();
+
+    String dumpTree();
+}

broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java

@@ -13,7 +13,6 @@
  *
  * You may elect to redistribute this code under either of these licenses.
  */
-
 package io.moquette.spi.impl.subscriptions;
 
 import io.moquette.spi.ISessionsStore;
@@ -28,14 +27,14 @@ import java.util.concurrent.atomic.AtomicReference;
 /**
  * Represents a tree of topics subscriptions.
  */
-public class SubscriptionsDirectory {
+public class SubscriptionsDirectory implements ISubscriptionsDirectory {
 
-    public static class NodeCouple {
+    static class NodeCouple {
 
         final TreeNode root;
         final TreeNode createdNode;
 
-        public NodeCouple(TreeNode root, TreeNode createdNode) {
+        NodeCouple(TreeNode root, TreeNode createdNode) {
             this.root = root;
             this.createdNode = createdNode;
         }
@@ -54,10 +53,10 @@ public class SubscriptionsDirectory {
 
         @Override
         public void visit(TreeNode node, int deep) {
-            String subScriptionsStr = "";
+            StringBuilder subScriptionsStr = new StringBuilder();
             String indentTabs = indentTabs(deep);
             for (ClientTopicCouple couple : node.m_subscriptions) {
-                subScriptionsStr += indentTabs + couple.toString() + "\n";
+                subScriptionsStr.append(indentTabs).append(couple.toString()).append("\n");
             }
             s += node.getToken() == null ? "" : node.getToken().toString();
             s += "\n" + (node.m_subscriptions.isEmpty() ? indentTabs : "")
@@ -65,12 +64,12 @@ public class SubscriptionsDirectory {
         }
 
         private String indentTabs(int deep) {
-            String s = "";
+            StringBuilder s = new StringBuilder();
             for (int i = 0; i < deep; i++) {
-                s += "\t";
+                s.append("\t");
                 // s += "--";
             }
-            return s;
+            return s.toString();
         }
 
         @Override
@@ -81,7 +80,6 @@ public class SubscriptionsDirectory {
 
     private AtomicReference<TreeNode> subscriptions = new AtomicReference<>(new TreeNode());
     private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsDirectory.class);
-    private volatile ISessionsStore m_sessionsStore;
     private volatile ISubscriptionsStore subscriptionsStore;
 
     /**
@@ -91,9 +89,9 @@ public class SubscriptionsDirectory {
      * @param sessionsStore
      *            to be used as backing store from the subscription store.
      */
+    @Override
     public void init(ISessionsStore sessionsStore) {
         LOG.info("Initializing subscriptions store...");
-        m_sessionsStore = sessionsStore;
         this.subscriptionsStore = sessionsStore.subscriptionStore();
         List<ClientTopicCouple> subscriptions = this.subscriptionsStore.listAllSubscriptions();
         // reload any subscriptions persisted
@@ -111,6 +109,7 @@ public class SubscriptionsDirectory {
         }
     }
 
+    @Override
     public void add(ClientTopicCouple newSubscription) {
         /*
          * The topic filters have already been validated at the ProtocolProcessor. We can assume
@@ -128,7 +127,7 @@ public class SubscriptionsDirectory {
         LOG.debug("A subscription has been added. Root = {}, oldRoot = {}.", couple.root, oldRoot);
     }
 
-    protected NodeCouple recreatePath(Topic topic, final TreeNode oldRoot) {
+    NodeCouple recreatePath(Topic topic, final TreeNode oldRoot) {
         final TreeNode newRoot = oldRoot.copy();
         TreeNode parent = newRoot;
         TreeNode current = newRoot;
@@ -153,6 +152,7 @@ public class SubscriptionsDirectory {
         return new NodeCouple(newRoot, current);
     }
 
+    @Override
     public void removeSubscription(Topic topic, String clientID) {
         /*
          * The topic filters have already been validated at the ProtocolProcessor. We can assume
@@ -170,22 +170,22 @@ public class SubscriptionsDirectory {
         } while (!subscriptions.compareAndSet(oldRoot, couple.root));
     }
 
-    /**
-     * Visit the topics tree to remove matching subscriptions with clientID. It's a mutating
-     * structure operation so create a new subscription tree (partial or total).
-     *
-     * @param clientID
-     *            the client ID to remove.
-     */
-    public void removeForClient(String clientID) {
-        TreeNode oldRoot;
-        TreeNode newRoot;
-        do {
-            oldRoot = subscriptions.get();
-            newRoot = oldRoot.removeClientSubscriptions(clientID);
-            // spin lock repeating till we can, swap root, if can't swap just re-do the operation
-        } while (!subscriptions.compareAndSet(oldRoot, newRoot));
-    }
+//    /**
+//     * Visit the topics tree to remove matching subscriptions with clientID. It's a mutating
+//     * structure operation so create a new subscription tree (partial or total).
+//     *
+//     * @param clientID
+//     *            the client ID to remove.
+//     */
+//    void removeForClient(String clientID) {
+//        TreeNode oldRoot;
+//        TreeNode newRoot;
+//        do {
+//            oldRoot = subscriptions.get();
+//            newRoot = oldRoot.removeClientSubscriptions(clientID);
+//            // spin lock repeating till we can, swap root, if can't swap just re-do the operation
+//        } while (!subscriptions.compareAndSet(oldRoot, newRoot));
+//    }
 
     /**
      * Given a topic string return the clients subscriptions that matches it. Topic string can't
@@ -196,6 +196,7 @@ public class SubscriptionsDirectory {
      *            to use fo searching matching subscriptions.
      * @return the list of matching subscriptions, or empty if not matching.
      */
+    @Override
     public List<Subscription> matches(Topic topic) {
         Queue<Token> tokenQueue = new LinkedBlockingDeque<>(topic.getTokens());
         List<ClientTopicCouple> matchingSubs = new ArrayList<>();
@@ -218,10 +219,12 @@ public class SubscriptionsDirectory {
         return new ArrayList<>(subsForClient.values());
     }
 
+    @Override
     public int size() {
         return subscriptions.get().size();
     }
 
+    @Override
     public String dumpTree() {
         DumpTreeVisitor visitor = new DumpTreeVisitor();
         bfsVisit(subscriptions.get(), visitor, 0);

broker/src/main/java/io/moquette/spi/impl/subscriptions/TNode.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright (c) 2012-2017 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 io.moquette.spi.impl.subscriptions;
+
+class TNode extends CNode {
+
+    @Override
+    INode childOf(Token token) {
+        throw new IllegalStateException("Can't be invoked on TNode");
+    }
+
+    @Override
+    CNode copy() {
+        throw new IllegalStateException("Can't be invoked on TNode");
+    }
+
+    @Override
+    public void add(INode newINode) {
+        throw new IllegalStateException("Can't be invoked on TNode");
+    }
+
+    @Override
+    CNode addSubscription(String clientId, Topic topic) {
+        throw new IllegalStateException("Can't be invoked on TNode");
+    }
+
+    @Override
+    boolean containsOnly(String clientId) {
+        throw new IllegalStateException("Can't be invoked on TNode");
+    }
+
+    @Override
+    public boolean contains(String clientId) {
+        throw new IllegalStateException("Can't be invoked on TNode");
+    }
+
+    @Override
+    void removeSubscriptionsFor(String clientId) {
+        throw new IllegalStateException("Can't be invoked on TNode");
+    }
+
+    @Override
+    boolean anyChildrenMatch(Token token) {
+        return false;
+    }
+}

broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java

@@ -19,8 +19,11 @@ package io.moquette.spi.impl.subscriptions;
 import java.io.Serializable;
 import java.text.ParseException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.stream.Collectors;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,10 +39,24 @@ public class Topic implements Serializable {
 
     private transient boolean valid;
 
+    /**
+     * Factory method
+     * */
+    public static Topic asTopic(String s) {
+        return new Topic(s);
+    }
+
     public Topic(String topic) {
         this.topic = topic;
     }
 
+    Topic(List<Token> tokens) {
+        this.tokens = tokens;
+        List<String> strTokens = tokens.stream().map(Token::toString).collect(Collectors.toList());
+        this.topic = String.join("/", strTokens);
+        this.valid = true;
+    }
+
     public List<Token> getTokens() {
         if (tokens == null) {
             try {
@@ -100,6 +117,33 @@ public class Topic implements Serializable {
         return res;
     }
 
+    public Token headToken() {
+        final List<Token> tokens = getTokens();
+        if (tokens.isEmpty()) {
+            //TODO UGLY use Optional
+            return null;
+        }
+        return tokens.get(0);
+    }
+
+    public boolean isEmpty() {
+        final List<Token> tokens = getTokens();
+        return tokens == null || tokens.isEmpty();
+    }
+
+    /**
+     * @return a new Topic corresponding to this less than the head token
+     * */
+    public Topic exceptHeadToken() {
+        List<Token> tokens = getTokens();
+        if (tokens.isEmpty()) {
+            return new Topic(Collections.emptyList());
+        }
+        List<Token> tokensCopy = new ArrayList<>(tokens);
+        tokensCopy.remove(0);
+        return new Topic(tokensCopy);
+    }
+
     public boolean isValid() {
         if (tokens == null)
             getTokens();
@@ -168,10 +212,4 @@ public class Topic implements Serializable {
         return topic.hashCode();
     }
 
-    /**
-     * Factory method
-     * */
-    public static Topic asTopic(String s) {
-        return new Topic(s);
-    }
 }