diff --git a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java b/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java index a10b3d8d..5a4908cd 100644 --- a/broker/src/main/java/io/moquette/spi/impl/MessagesPublisher.java +++ b/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; diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index c404566b..cd099cf6 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/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 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) { diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index 6531b7ab..44d1739f 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/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..."); diff --git a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java b/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java index dd1cc6a7..1490fae3 100644 --- a/broker/src/main/java/io/moquette/spi/impl/Qos2PublishHandler.java +++ b/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) { diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java new file mode 100644 index 00000000..c4322000 --- /dev/null +++ b/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 children; + Set subscriptions = new HashSet<>(); + + //private int subtreeSubscriptions; + + CNode() { + this.children = new ArrayList<>(); + this.subscriptions = new HashSet<>(); + } + + //Copy constructor + private CNode(Token token, List children, Set 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 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 toRemove = new HashSet<>(); + for (ClientTopicCouple sub : this.subscriptions) { + if (sub.clientID.equals(clientId)) { + toRemove.add(sub); + } + } + this.subscriptions.removeAll(toRemove); + } + +// public List childrenMatching(Token token) { +// List 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; +// } +} diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java new file mode 100644 index 00000000..9f349a36 --- /dev/null +++ b/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 { + + void visit(CNode node, int deep); + + T getResult(); + } + + private class DumpTreeVisitor implements IVisitor { + + 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 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 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 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 match(Topic topic) { + final Set matchingSubs = recursiveMatch(topic, this.root); + // remove the overlapping subscriptions, selecting ones with greatest qos + Map 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 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 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 { + + 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); + } + } +} diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/INode.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/INode.java new file mode 100644 index 00000000..0dbeec59 --- /dev/null +++ b/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 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(); + } +} diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/ISubscriptionsDirectory.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/ISubscriptionsDirectory.java new file mode 100644 index 00000000..3452ba8e --- /dev/null +++ b/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 matches(Topic topic); + + int size(); + + String dumpTree(); +} diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java index c2e5cdd1..bc3c1aa6 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java +++ b/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 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 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 matches(Topic topic) { Queue tokenQueue = new LinkedBlockingDeque<>(topic.getTokens()); List 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); diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/TNode.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/TNode.java new file mode 100644 index 00000000..11f0995f --- /dev/null +++ b/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; + } +} diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java index 51f8dd58..8ed70a95 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Topic.java +++ b/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 tokens) { + this.tokens = tokens; + List strTokens = tokens.stream().map(Token::toString).collect(Collectors.toList()); + this.topic = String.join("/", strTokens); + this.valid = true; + } + public List getTokens() { if (tokens == null) { try { @@ -100,6 +117,33 @@ public class Topic implements Serializable { return res; } + public Token headToken() { + final List tokens = getTokens(); + if (tokens.isEmpty()) { + //TODO UGLY use Optional + return null; + } + return tokens.get(0); + } + + public boolean isEmpty() { + final List tokens = getTokens(); + return tokens == null || tokens.isEmpty(); + } + + /** + * @return a new Topic corresponding to this less than the head token + * */ + public Topic exceptHeadToken() { + List tokens = getTokens(); + if (tokens.isEmpty()) { + return new Topic(Collections.emptyList()); + } + List 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); - } } diff --git a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java index 0d895b15..22fe2734 100644 --- a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java +++ b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java @@ -17,9 +17,7 @@ package io.moquette.spi; import io.moquette.persistence.MemoryStorageService; -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.impl.subscriptions.*; import org.junit.Before; import org.junit.Test; @@ -34,11 +32,11 @@ public class ClientSessionTest { ClientSession session1; ClientSession session2; ISessionsStore sessionsStore; - SubscriptionsDirectory store; + ISubscriptionsDirectory store; @Before public void setUp() { - store = new SubscriptionsDirectory(); + store = new CTrieSubscriptionDirectory(); MemoryStorageService storageService = new MemoryStorageService(null, null); this.sessionsStore = storageService.sessionsStore(); store.init(sessionsStore); diff --git a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java index 44f51698..d9e5f24f 100644 --- a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java +++ b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java @@ -22,9 +22,7 @@ import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; import io.moquette.spi.impl.security.PermitAllAuthorizator; -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.impl.subscriptions.*; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.embedded.EmbeddedChannel; @@ -56,7 +54,7 @@ abstract class AbstractProtocolProcessorCommonUtils { IMessagesStore m_messagesStore; ISessionsStore m_sessionStore; - SubscriptionsDirectory subscriptions; + ISubscriptionsDirectory subscriptions; MockAuthenticator m_mockAuthenticator; protected void initializeProcessorAndSubsystems() { @@ -77,7 +75,7 @@ abstract class AbstractProtocolProcessorCommonUtils { users.put(TEST_USER, TEST_PWD); m_mockAuthenticator = new MockAuthenticator(clientIds, users); - subscriptions = new SubscriptionsDirectory(); + subscriptions = new CTrieSubscriptionDirectory(); subscriptions.init(memStorage.sessionsStore()); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 86efb16b..2589d1ac 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -22,6 +22,7 @@ import io.moquette.server.netty.NettyUtils; import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore.StoredMessage; import io.moquette.spi.impl.security.PermitAllAuthorizator; +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; @@ -68,7 +69,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true - SubscriptionsDirectory subs = new SubscriptionsDirectory() { + ISubscriptionsDirectory subs = new SubscriptionsDirectory() { @Override public List matches(Topic topic) { @@ -104,7 +105,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true - SubscriptionsDirectory subs = new SubscriptionsDirectory() { + ISubscriptionsDirectory subs = new SubscriptionsDirectory() { @Override public List matches(Topic topic) { @@ -228,7 +229,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils new Subscription(FAKE_PUBLISHER_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true - SubscriptionsDirectory subs = new SubscriptionsDirectory() { + ISubscriptionsDirectory subs = new SubscriptionsDirectory() { @Override public List matches(Topic topic) { @@ -258,7 +259,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils @Test public void testRepublishAndConsumePersistedMessages_onReconnect() { - SubscriptionsDirectory subs = mock(SubscriptionsDirectory.class); + ISubscriptionsDirectory subs = mock(SubscriptionsDirectory.class); List emptySubs = Collections.emptyList(); when(subs.matches(any(Topic.class))).thenReturn(emptySubs); @@ -282,7 +283,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils // create an inactive session for Subscriber m_sessionStore.createNewSession("Subscriber", false); - SubscriptionsDirectory mockedSubscriptions = mock(SubscriptionsDirectory.class); + ISubscriptionsDirectory mockedSubscriptions = mock(SubscriptionsDirectory.class); Subscription inactiveSub = new Subscription("Subscriber", new Topic("/topic"), MqttQoS.AT_LEAST_ONCE); List inactiveSubscriptions = Collections.singletonList(inactiveSub); when(mockedSubscriptions.matches(eq(new Topic("/topic")))).thenReturn(inactiveSubscriptions); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index d51d9eed..77728487 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -22,9 +22,7 @@ import io.moquette.spi.ClientSession; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.security.PermitAllAuthorizator; -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.impl.subscriptions.*; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.mqtt.*; import org.junit.Before; @@ -47,7 +45,7 @@ public class ProtocolProcessor_CONNECT_Test { IMessagesStore m_messagesStore; ISessionsStore m_sessionStore; - SubscriptionsDirectory subscriptions; + ISubscriptionsDirectory subscriptions; MockAuthenticator m_mockAuthenticator; @Before @@ -63,7 +61,7 @@ public class ProtocolProcessor_CONNECT_Test { m_mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD)); - subscriptions = new SubscriptionsDirectory(); + subscriptions = new CTrieSubscriptionDirectory(); subscriptions.init(m_sessionStore); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java new file mode 100644 index 00000000..e4ee57cf --- /dev/null +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java @@ -0,0 +1,285 @@ +/* + * 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.persistence.MemoryStorageService; +import io.moquette.spi.ClientSession; +import io.moquette.spi.ISessionsStore; +import io.moquette.spi.ISubscriptionsStore; +import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; +import io.netty.handler.codec.mqtt.MqttQoS; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +import java.util.List; +import java.util.Optional; +import java.util.Set; + +import static io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectoryTest.clientSubOnTopic; +import static io.moquette.spi.impl.subscriptions.Topic.asTopic; +import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +public class CTrieSubscriptionDirectoryMatchingTest { + + private CTrieSubscriptionDirectory sut; + private ISessionsStore sessionsStore; + private ISubscriptionsStore subscriptionsStore; + private MemoryStorageService storageService; + + @Before + public void setUp() { + sut = new CTrieSubscriptionDirectory(); + + this.storageService = new MemoryStorageService(null, null); + this.sessionsStore = storageService.sessionsStore(); + sut.init(sessionsStore); + + this.subscriptionsStore = this.sessionsStore.subscriptionStore(); + } + + @Test + public void testMatchSimple() { + ClientTopicCouple slashSub = clientSubOnTopic("TempSensor1", "/"); + sut.add(slashSub); + assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).isEmpty(); + + ClientTopicCouple slashFinanceSub = clientSubOnTopic("TempSensor1", "/finance"); + sut.add(slashFinanceSub); + assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).isEmpty(); + + assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).contains(slashFinanceSub); + assertThat(sut.recursiveMatch(asTopic("/"), sut.root)).contains(slashSub); + } + + @Test + public void testMatchSimpleMulti() { + ClientTopicCouple anySub = clientSubOnTopic("TempSensor1", "#"); + sut.add(anySub); + assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).contains(anySub); + + ClientTopicCouple financeAnySub = clientSubOnTopic("TempSensor1", "finance/#"); + sut.add(financeAnySub); + assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).containsExactlyInAnyOrder(financeAnySub, anySub); + } + + @Test + public void testMatchingDeepMulti_one_layer() { + ClientTopicCouple anySub = clientSubOnTopic("AllSensor1", "#"); + ClientTopicCouple financeAnySub = clientSubOnTopic("FinanceSensor", "finance/#"); + sut.add(anySub); + sut.add(financeAnySub); + + // Verify + assertThat(sut.recursiveMatch(asTopic("finance/stock"), sut.root)).containsExactlyInAnyOrder(financeAnySub, anySub); + assertThat(sut.recursiveMatch(asTopic("finance/stock/ibm"), sut.root)).containsExactlyInAnyOrder(financeAnySub, anySub); + + System.out.println(sut.dumpTree()); + } + + @Test + public void testMatchingDeepMulti_two_layer() { + ClientTopicCouple financeAnySub = clientSubOnTopic("FinanceSensor", "finance/stock/#"); + sut.add(financeAnySub); + + // Verify + assertThat(sut.recursiveMatch(asTopic("finance/stock/ibm"), sut.root)).containsExactly(financeAnySub); + } + + @Test + public void testMatchSimpleSingle() { + ClientTopicCouple anySub = clientSubOnTopic("AnySensor", "+"); + sut.add(anySub); + assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).containsExactly(anySub); + + ClientTopicCouple financeOne = clientSubOnTopic("AnySensor", "finance/+"); + sut.add(financeOne); + assertThat(sut.recursiveMatch(asTopic("finance/stock"), sut.root)).containsExactly(financeOne); + } + + @Test + public void testMatchManySingle() { + ClientTopicCouple manySub = clientSubOnTopic("AnySensor", "+/+"); + sut.add(manySub); + + // verify + assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).contains(manySub); + } + + @Test + public void testMatchSlashSingle() { + ClientTopicCouple slashPlusSub = clientSubOnTopic("AnySensor", "/+"); + sut.add(slashPlusSub); + ClientTopicCouple anySub = clientSubOnTopic("AnySensor", "+"); + sut.add(anySub); + + // Verify + assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).containsOnly(slashPlusSub); + assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).doesNotContain(anySub); + } + + @Test + public void testMatchManyDeepSingle() { + ClientTopicCouple slashPlusSub = clientSubOnTopic("FinanceSensor1", "/finance/+/ibm"); + sut.add(slashPlusSub); + ClientTopicCouple slashPlusDeepSub = clientSubOnTopic("FinanceSensor2", "/+/stock/+"); + sut.add(slashPlusDeepSub); + + // Verify + assertThat(sut.recursiveMatch(asTopic("/finance/stock/ibm"), sut.root)) + .containsExactlyInAnyOrder(slashPlusSub, slashPlusDeepSub); + } + + @Test + public void testMatchSimpleMulti_allTheTree() { + ClientTopicCouple sub = clientSubOnTopic("AnySensor1", "#"); + sut.add(sub); + + assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).isNotEmpty(); + assertThat(sut.recursiveMatch(asTopic("finance/ibm"), sut.root)).isNotEmpty(); + } + + @Test + public void rogerLightTopicMatches() { + assertMatch("foo/bar", "foo/bar"); + assertMatch("foo/bar", "foo/bar"); + assertMatch("foo/+", "foo/bar"); + assertMatch("foo/+/baz", "foo/bar/baz"); + assertMatch("foo/+/#", "foo/bar/baz"); + assertMatch("#", "foo/bar/baz"); + + assertNotMatch("foo/bar", "foo"); + assertNotMatch("foo/+", "foo/bar/baz"); + assertNotMatch("foo/+/baz", "foo/bar/bar"); + assertNotMatch("foo/+/#", "fo2/bar/baz"); + + assertMatch("#", "/foo/bar"); + assertMatch("/#", "/foo/bar"); + assertNotMatch("/#", "foo/bar"); + + assertMatch("foo//bar", "foo//bar"); + assertMatch("foo//+", "foo//bar"); + assertMatch("foo/+/+/baz", "foo///baz"); + assertMatch("foo/bar/+", "foo/bar/"); + } + + private void assertMatch(String s, String t) { + sut = new CTrieSubscriptionDirectory(); + MemoryStorageService memStore = new MemoryStorageService(null, null); + ISessionsStore aSessionsStore = memStore.sessionsStore(); + sut.init(aSessionsStore); + + ClientTopicCouple sub = clientSubOnTopic("AnySensor1", s); + sut.add(sub); + + assertThat(sut.recursiveMatch(asTopic(t), sut.root)).isNotEmpty(); + } + + private void assertNotMatch(String subscription, String topic) { + sut = new CTrieSubscriptionDirectory(); + MemoryStorageService memStore = new MemoryStorageService(null, null); + ISessionsStore aSessionsStore = memStore.sessionsStore(); + sut.init(aSessionsStore); + + ClientTopicCouple sub = clientSubOnTopic("AnySensor1", subscription); + sut.add(sub); + + assertThat(sut.recursiveMatch(asTopic(topic), sut.root)).isEmpty(); + } + + @Test + public void testOverlappingSubscriptions() { + Subscription sub = new Subscription("Sensor1", asTopic("a/+"), MqttQoS.AT_MOST_ONCE); + ClientTopicCouple genericSub = sub.asClientTopicCouple(); + this.subscriptionsStore.addNewSubscription(sub); + storageService.sessionsStore().createNewSession("Sensor1", false); + sut.add(genericSub); + + Subscription sub2 = new Subscription("Sensor1", asTopic("a/b"), MqttQoS.AT_MOST_ONCE); + ClientTopicCouple specificSub = sub2.asClientTopicCouple(); + this.subscriptionsStore.addNewSubscription(sub2); + sut.add(specificSub); + + // Verify + assertThat(sut.match(asTopic("a/b")).size()).isEqualTo(1); + } + + @Test + public void removeSubscription_withDifferentClients_subscribedSameTopic() { + ClientTopicCouple slashSub = clientSubOnTopic("Sensor1", "/topic"); + sut.add(slashSub); + ClientTopicCouple slashSub2 = clientSubOnTopic("Sensor2", "/topic"); + sut.add(slashSub2); + + // Exercise + sut.removeSubscription(asTopic("/topic"), slashSub2.clientID); + + // Verify + ClientTopicCouple remainedSubscription = sut.recursiveMatch(asTopic("/topic"), sut.root).iterator().next(); + assertThat(remainedSubscription.clientID).isEqualTo(slashSub.clientID); + assertEquals(slashSub.clientID, remainedSubscription.clientID); + } + + @Test + public void removeSubscription_sameClients_subscribedSameTopic() { + ClientTopicCouple slashSub = clientSubOnTopic("Sensor1", "/topic"); + sut.add(slashSub); + + // Exercise + sut.removeSubscription(asTopic("/topic"), slashSub.clientID); + + // Verify + final Set matchingSubscriptions = sut.recursiveMatch(asTopic("/topic"), sut.root); + assertThat(matchingSubscriptions).isEmpty(); + } + + /* + * Test for Issue #49 + */ + @Test + public void duplicatedSubscriptionsWithDifferentQos() { + ClientSession session2 = sessionsStore.createNewSession("client2", true); + Subscription client2Sub = new Subscription("client2", asTopic("client/test/b"), MqttQoS.AT_MOST_ONCE); + session2.subscribe(client2Sub); + this.sut.add(client2Sub.asClientTopicCouple()); + ClientSession session1 = sessionsStore.createNewSession("client1", true); + Subscription client1SubQoS0 = new Subscription("client1", asTopic("client/test/b"), MqttQoS.AT_MOST_ONCE); + session1.subscribe(client1SubQoS0); + this.sut.add(client1SubQoS0.asClientTopicCouple()); + + Subscription client1SubQoS2 = new Subscription("client1", asTopic("client/test/b"), MqttQoS.EXACTLY_ONCE); + session1.subscribe(client1SubQoS2); + this.sut.add(client1SubQoS2.asClientTopicCouple()); + + // Verify + Set subscriptions = this.sut.match(asTopic("client/test/b")); + assertThat(subscriptions).contains(client1SubQoS2); + assertThat(subscriptions).contains(client2Sub); + + final Optional matchingClient1Sub = subscriptions + .stream() + .filter(s -> s.equals(client1SubQoS0)) + .findFirst(); + assertTrue(matchingClient1Sub.isPresent()); + Subscription client1Sub = matchingClient1Sub.get(); + + assertThat(client1SubQoS0.getRequestedQos()).isNotEqualTo(client1Sub.getRequestedQos()); + + // client1SubQoS2 should override client1SubQoS0 + assertThat(client1Sub.getRequestedQos()).isEqualTo(client1SubQoS2.getRequestedQos()); + } +} diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java new file mode 100644 index 00000000..f9f39e11 --- /dev/null +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java @@ -0,0 +1,144 @@ +/* + * 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.persistence.MemoryStorageService; +import io.moquette.spi.ISessionsStore; +import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; +import org.junit.Before; +import org.junit.Test; + +import java.util.Optional; +import java.util.Set; + +import static io.moquette.spi.impl.subscriptions.Topic.asTopic; +import static org.junit.Assert.*; +import static org.assertj.core.api.Assertions.assertThat; + +public class CTrieSubscriptionDirectoryTest { + + private CTrieSubscriptionDirectory sut; + + @Before + public void setUp() { + sut = new CTrieSubscriptionDirectory(); + MemoryStorageService memStore = new MemoryStorageService(null, null); + ISessionsStore aSessionsStore = memStore.sessionsStore(); + sut.init(aSessionsStore); + } + + @Test + public void testAddOnSecondLayerWithEmptyTokenOnEmptyTree() { + //Exercise + sut.add(clientSubOnTopic("TempSensor1", "/")); + + //Verify + final Optional matchedNode = sut.lookup(asTopic("/")); + assertTrue("Node on path / must be present", matchedNode.isPresent()); + //verify structure, only root INode and the first CNode should be present + assertThat(this.sut.root.mainNode().subscriptions).isEmpty(); + assertThat(this.sut.root.mainNode().allChildren()).isNotEmpty(); + + INode firstLayer = this.sut.root.mainNode().allChildren().get(0); + assertThat(firstLayer.mainNode().subscriptions).isEmpty(); + assertThat(firstLayer.mainNode().allChildren()).isNotEmpty(); + + INode secondLayer = firstLayer.mainNode().allChildren().get(0); + assertThat(secondLayer.mainNode().subscriptions).isNotEmpty(); + assertThat(secondLayer.mainNode().allChildren()).isEmpty(); + } + + @Test + public void testAddFirstLayerNodeOnEmptyTree() { + //Exercise + sut.add(clientSubOnTopic("TempSensor1", "/temp")); + + //Verify + final Optional matchedNode = sut.lookup(asTopic("/temp")); + assertTrue("Node on path /temp must be present", matchedNode.isPresent()); + assertFalse(matchedNode.get().subscriptions.isEmpty()); + } + + @Test + public void testLookup() { + final ClientTopicCouple existingSubscription = clientSubOnTopic("TempSensor1", "/temp"); + sut.add(existingSubscription); + + //Exercise + final Optional matchedNode = sut.lookup(asTopic("/humidity")); + + //Verify + assertFalse("Node on path /humidity can't be present", matchedNode.isPresent()); + } + + @Test + public void testAddNewSubscriptionOnExistingNode() { + final ClientTopicCouple existingSubscription = clientSubOnTopic("TempSensor1", "/temp"); + sut.add(existingSubscription); + + //Exercise + final ClientTopicCouple newSubscription = clientSubOnTopic("TempSensor2", "/temp"); + sut.add(newSubscription); + + //Verify + final Optional matchedNode = sut.lookup(asTopic("/temp")); + assertTrue("Node on path /temp must be present", matchedNode.isPresent()); + final Set subscriptions = matchedNode.get().subscriptions; + assertTrue(subscriptions.contains(newSubscription)); + } + + @Test + public void testAddNewDeepNodes() { + sut.add(clientSubOnTopic("TempSensorRM", "/italy/roma/temp")); + sut.add(clientSubOnTopic("TempSensorFI", "/italy/firenze/temp")); + sut.add(clientSubOnTopic("HumSensorFI", "/italy/roma/humidity")); + final ClientTopicCouple happinessSensor = clientSubOnTopic("HappinessSensor", "/italy/happiness"); + sut.add(happinessSensor); + + //Verify + final Optional matchedNode = sut.lookup(asTopic("/italy/happiness")); + assertTrue("Node on path /italy/happiness must be present", matchedNode.isPresent()); + final Set subscriptions = matchedNode.get().subscriptions; + assertTrue(subscriptions.contains(happinessSensor)); + } + + static ClientTopicCouple clientSubOnTopic(String clientID, String topicName) { + return new ClientTopicCouple(clientID, asTopic(topicName)); + } + + @Test + public void givenTreeWithSomeNodeWhenRemoveContainedSubscriptionThenNodeIsUpdated() { + sut.add(clientSubOnTopic("TempSensor1", "/temp")); + + //Exercise + sut.removeSubscription(asTopic("/temp"), "TempSensor1"); + + //Verify + final Optional matchedNode = sut.lookup(asTopic("/temp")); + assertFalse("Node on path /temp can't be present", matchedNode.isPresent()); + } + + @Test + public void testMatchSubscriptionNoWildcards() { + sut.add(clientSubOnTopic("TempSensor1", "/temp")); + + //Exercise + final Set matchingSubs = sut.recursiveMatch(asTopic("/temp"), sut.root); + + //Verify + final ClientTopicCouple expectedMatchingsub = new ClientTopicCouple("TempSensor1", asTopic("/temp")); + assertThat(matchingSubs).contains(expectedMatchingsub); + } +} diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectoryTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectoryTest.java index f5eb82c2..4373b527 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectoryTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectoryTest.java @@ -234,34 +234,6 @@ public class SubscriptionsDirectoryTest { assertThat(store.matches(topic)).isEmpty(); } - @Test - public void testRemoveClientSubscriptions_existingClientID() { - String cliendID = "FAKE_CLID_1"; - Subscription sub = new Subscription(cliendID, new Topic("finance/#"), MqttQoS.AT_MOST_ONCE); - this.subscriptionsStore.addNewSubscription(sub); - store.add(sub.asClientTopicCouple()); - - // Exercise - store.removeForClient(cliendID); - - // Verify - assertThat(store.size()).isZero(); - } - - @Test - public void testRemoveClientSubscriptions_notexistingClientID() { - String clientID = "FAKE_CLID_1"; - Subscription s = new Subscription(clientID, new Topic("finance/#"), MqttQoS.AT_MOST_ONCE); - this.subscriptionsStore.addNewSubscription(s); - store.add(s.asClientTopicCouple()); - - // Exercise - store.removeForClient("FAKE_CLID_2"); - - // Verify - assertThat(store.size()).isEqualTo(1); - } - @Test public void testOverlappingSubscriptions() { Subscription genericSub = new Subscription("FAKE_CLI_ID_1", new Topic("a/+"), MqttQoS.EXACTLY_ONCE); @@ -277,7 +249,7 @@ public class SubscriptionsDirectoryTest { @Test public void removeSubscription_withDifferentClients_subscribedSameTopic() { - SubscriptionsDirectory aStore = new SubscriptionsDirectory(); + ISubscriptionsDirectory aStore = new SubscriptionsDirectory(); MemoryStorageService memStore = new MemoryStorageService(null, null); ISessionsStore sessionsStore = memStore.sessionsStore(); aStore.init(sessionsStore); diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java index 94bf2005..95cfe5c2 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/TopicTest.java @@ -19,6 +19,8 @@ import org.assertj.core.api.AbstractAssert; import org.assertj.core.api.Assertions; import org.junit.Test; +import static org.junit.Assert.assertEquals; + public class TopicTest { @Test @@ -38,7 +40,7 @@ public class TopicTest { } @Test - public void testValidtionProcess() { + public void testValidationProcess() { // TopicMultiInTheMiddle assertThat(new Topic("finance/#/closingprice")).isInValid(); @@ -114,6 +116,13 @@ public class TopicTest { assertThat(new Topic("foo/bar/")).matches("foo/bar/+"); } + @Test + public void exceptHeadToken() { + assertEquals(Topic.asTopic("token"), Topic.asTopic("/token").exceptHeadToken()); + assertEquals(Topic.asTopic("a/b"), Topic.asTopic("/a/b").exceptHeadToken()); + } + + public static TopicAssert assertThat(Topic topic) { return new TopicAssert(topic); } diff --git a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java index bec33620..d2365786 100644 --- a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java +++ b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java @@ -34,7 +34,7 @@ public final class PerformanceTest { cpufreq-set -g performance -u 800MHz -c 3 */ public static void main(String[] args) throws IOException, InterruptedException { - SubscriptionsDirectory store = new SubscriptionsDirectory(); + ISubscriptionsDirectory store = new SubscriptionsDirectory(); MemoryStorageService memStore = new MemoryStorageService(null, null); ISessionsStore aSessionsStore = memStore.sessionsStore(); store.init(aSessionsStore);