diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/CTrie.java b/broker/src/main/java/io/moquette/broker/subscriptions/CTrie.java new file mode 100644 index 00000000..70e279f0 --- /dev/null +++ b/broker/src/main/java/io/moquette/broker/subscriptions/CTrie.java @@ -0,0 +1,233 @@ +package io.moquette.broker.subscriptions; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Optional; +import java.util.Set; + +public class CTrie { + + interface IVisitor { + + void visit(CNode node, int deep); + + T getResult(); + } + + private static final Token ROOT = new Token("root"); + private static final INode NO_PARENT = null; + + private enum Action { + OK, REPEAT + } + + INode root; + + CTrie() { + final CNode mainNode = new CNode(); + mainNode.token = ROOT; + this.root = new INode(mainNode); + } + + 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()); + } + + enum NavigationAction { + MATCH, GODEEP, STOP + } + + private NavigationAction evaluate(Topic topic, CNode cnode) { + if (Token.MULTI.equals(cnode.token)) { + return NavigationAction.MATCH; + } + if (topic.isEmpty()) { + return NavigationAction.STOP; + } + final Token token = topic.headToken(); + if (!(Token.SINGLE.equals(cnode.token) || cnode.token.equals(token) || ROOT.equals(cnode.token))) { + return NavigationAction.STOP; + } + return NavigationAction.GODEEP; + } + + public Set recursiveMatch(Topic topic) { + return recursiveMatch(topic, this.root); + } + + private Set recursiveMatch(Topic topic, INode inode) { + CNode cnode = inode.mainNode(); + NavigationAction action = evaluate(topic, cnode); + if (action == NavigationAction.MATCH) { + return cnode.subscriptions; + } + if (action == NavigationAction.STOP) { + return Collections.emptySet(); + } + if (cnode instanceof TNode) { + return Collections.emptySet(); + } + Topic remainingTopic = (ROOT.equals(cnode.token)) ? 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 addToTree(Subscription newSubscription) { + Action res; + do { + res = insert(newSubscription.topicFilter, this.root, newSubscription); + } while (res == Action.REPEAT); + } + + private Action insert(Topic topic, final INode inode, Subscription newSubscription) { + Token token = topic.headToken(); + if (!topic.isEmpty() && inode.mainNode().anyChildrenMatch(token)) { + Topic remainingTopic = topic.exceptHeadToken(); + INode nextInode = inode.mainNode().childOf(token); + return insert(remainingTopic, nextInode, newSubscription); + } else { + if (topic.isEmpty()) { + return insertSubscription(inode, newSubscription); + } else { + return createNodeAndInsertSubscription(topic, inode, newSubscription); + } + } + } + + private Action insertSubscription(INode inode, Subscription newSubscription) { + CNode cnode = inode.mainNode(); + CNode updatedCnode = cnode.copy().addSubscription(newSubscription); + if (inode.compareAndSet(cnode, updatedCnode)) { + return Action.OK; + } else { + return Action.REPEAT; + } + } + + private Action createNodeAndInsertSubscription(Topic topic, INode inode, Subscription newSubscription) { + INode newInode = createPathRec(topic, newSubscription); + CNode cnode = inode.mainNode(); + CNode updatedCnode = cnode.copy(); + updatedCnode.add(newInode); + + return inode.compareAndSet(cnode, updatedCnode) ? Action.OK : Action.REPEAT; + } + + private INode createPathRec(Topic topic, Subscription newSubscription) { + Topic remainingTopic = topic.exceptHeadToken(); + if (!remainingTopic.isEmpty()) { + INode inode = createPathRec(remainingTopic, newSubscription); + CNode cnode = new CNode(); + cnode.token = topic.headToken(); + cnode.add(inode); + return new INode(cnode); + } else { + return createLeafNodes(topic.headToken(), newSubscription); + } + } + + private INode createLeafNodes(Token token, Subscription newSubscription) { + CNode newLeafCnode = new CNode(); + newLeafCnode.token = token; + newLeafCnode.addSubscription(newSubscription); + + return new INode(newLeafCnode); + } + + public void removeFromTree(Topic topic, String clientID) { + Action res; + do { + res = remove(clientID, topic, this.root, NO_PARENT); + } while (res == Action.REPEAT); + } + + private Action remove(String clientId, Topic topic, INode inode, INode iParent) { + 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, inode); + } else { + final CNode cnode = inode.mainNode(); + if (cnode instanceof TNode) { + // this inode is a tomb, has no clients and should be cleaned up + // Because we implemented cleanTomb below, this should be rare, but possible + // Consider calling cleanTomb here too + return Action.OK; + } + if (cnode.containsOnly(clientId) && topic.isEmpty() && cnode.allChildren().isEmpty()) { + // last client to leave this node, AND there are no downstream children, remove via TNode tomb + if (inode == this.root) { + return inode.compareAndSet(cnode, inode.mainNode().copy()) ? Action.OK : Action.REPEAT; + } + TNode tnode = new TNode(); + return inode.compareAndSet(cnode, tnode) ? cleanTomb(inode, iParent) : Action.REPEAT; + } else if (cnode.contains(clientId) && topic.isEmpty()) { + CNode updatedCnode = cnode.copy(); + updatedCnode.removeSubscriptionsFor(clientId); + return inode.compareAndSet(cnode, updatedCnode) ? Action.OK : Action.REPEAT; + } else { + //someone else already removed + return Action.OK; + } + } + } + + /** + * + * Cleans Disposes of TNode in separate Atomic CAS operation per + * http://bravenewgeek.com/breaking-and-entering-lose-the-lock-while-embracing-concurrency/ + * + * We roughly follow this theory above, but we allow CNode with no Subscriptions to linger (for now). + * + * + * @param inode inode that handle to the tomb node. + * @param iParent inode parent. + * @return REPEAT if the this methods wasn't successful or OK. + */ + private Action cleanTomb(INode inode, INode iParent) { + CNode updatedCnode = iParent.mainNode().copy(); + updatedCnode.remove(inode); + return iParent.compareAndSet(iParent.mainNode(), updatedCnode) ? Action.OK : Action.REPEAT; + } + + 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/broker/subscriptions/CTrieSubscriptionDirectory.java b/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java index aa8a5f35..3dc71c19 100644 --- a/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java +++ b/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java @@ -25,29 +25,13 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { private static final Logger LOG = LoggerFactory.getLogger(CTrieSubscriptionDirectory.class); - private static final Token ROOT = new Token("root"); - private static final INode NO_PARENT = null; - - INode root; + private CTrie ctrie; private volatile ISubscriptionsRepository subscriptionsRepository; - interface IVisitor { - - void visit(CNode node, int deep); - - T getResult(); - } - - private enum Action { - OK, REPEAT - } - @Override public void init(ISubscriptionsRepository subscriptionsRepository) { LOG.info("Initializing CTrie"); - final CNode mainNode = new CNode(); - mainNode.token = ROOT; - this.root = new INode(mainNode); + ctrie = new CTrie(); LOG.info("Initializing subscriptions store..."); this.subscriptionsRepository = subscriptionsRepository; @@ -58,7 +42,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { for (Subscription subscription : this.subscriptionsRepository.listAllSubscriptions()) { LOG.debug("Re-subscribing {}", subscription); - addToTree(subscription); + ctrie.addToTree(subscription); } if (LOG.isTraceEnabled()) { LOG.trace("Stored subscriptions have been reloaded. SubscriptionTree = {}", dumpTree()); @@ -66,17 +50,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { } 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()); + return ctrie.lookup(topic); } /** @@ -90,7 +64,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { */ @Override public Set matchWithoutQosSharpening(Topic topic) { - return recursiveMatch(topic, this.root); + return ctrie.recursiveMatch(topic); } @Override @@ -108,118 +82,12 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { return new HashSet<>(subsGroupedByClient.values()); } - Set recursiveMatch(Topic topic, INode inode) { - CNode cnode = inode.mainNode(); - if (Token.MULTI.equals(cnode.token)) { - return cnode.subscriptions; - } - if (topic.isEmpty()) { - return Collections.emptySet(); - } - if (cnode instanceof TNode) { - return Collections.emptySet(); - } - final Token token = topic.headToken(); - if (!(Token.SINGLE.equals(cnode.token) || cnode.token.equals(token) || ROOT.equals(cnode.token))) { - return Collections.emptySet(); - } - Topic remainingTopic = (ROOT.equals(cnode.token)) ? 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; - } - - /** - * - * Cleans Disposes of TNode in separate Atomic CAS operation per - * http://bravenewgeek.com/breaking-and-entering-lose-the-lock-while-embracing-concurrency/ - * - * We roughly follow this theory above, but we allow CNode with no Subscriptions to linger (for now). - * - * - * @param inode inode that handle to the tomb node. - * @param iParent inode parent. - * @return REPEAT if the this methods wasn't successful or OK. - */ - public Action cleanTomb(INode inode, INode iParent) { - CNode updatedCnode = iParent.mainNode().copy(); - updatedCnode.remove(inode); - return iParent.compareAndSet(iParent.mainNode(), updatedCnode) ? Action.OK : Action.REPEAT; - } - @Override public void add(Subscription newSubscription) { - addToTree(newSubscription); + ctrie.addToTree(newSubscription); subscriptionsRepository.addNewSubscription(newSubscription); } - private void addToTree(Subscription newSubscription) { - Action res; - do { - res = insert(newSubscription.topicFilter, this.root, newSubscription); - } while (res == Action.REPEAT); - } - - private Action insert(Topic topic, final INode inode, Subscription newSubscription) { - Token token = topic.headToken(); - if (!topic.isEmpty() && inode.mainNode().anyChildrenMatch(token)) { - Topic remainingTopic = topic.exceptHeadToken(); - INode nextInode = inode.mainNode().childOf(token); - return insert(remainingTopic, nextInode, newSubscription); - } else { - if (topic.isEmpty()) { - return insertSubscription(inode, newSubscription); - } else { - return createNodeAndInsertSubscription(topic, inode, newSubscription); - } - } - } - - private Action insertSubscription(INode inode, Subscription newSubscription) { - CNode cnode = inode.mainNode(); - CNode updatedCnode = cnode.copy().addSubscription(newSubscription); - if (inode.compareAndSet(cnode, updatedCnode)) { - return Action.OK; - } else { - return Action.REPEAT; - } - } - - private Action createNodeAndInsertSubscription(Topic topic, INode inode, Subscription newSubscription) { - INode newInode = createPathRec(topic, newSubscription); - CNode cnode = inode.mainNode(); - CNode updatedCnode = cnode.copy(); - updatedCnode.add(newInode); - - return inode.compareAndSet(cnode, updatedCnode) ? Action.OK : Action.REPEAT; - } - - private INode createLeafNodes(Token token, Subscription newSubscription) { - CNode newLeafCnode = new CNode(); - newLeafCnode.token = token; - newLeafCnode.addSubscription(newSubscription); - - return new INode(newLeafCnode); - } - - private INode createPathRec(Topic topic, Subscription newSubscription) { - Topic remainingTopic = topic.exceptHeadToken(); - if (!remainingTopic.isEmpty()) { - INode inode = createPathRec(remainingTopic, newSubscription); - CNode cnode = new CNode(); - cnode.token = topic.headToken(); - cnode.add(inode); - return new INode(cnode); - } else { - return createLeafNodes(topic.headToken(), newSubscription); - } - } - /** * Removes subscription from CTrie, adds TNode when the last client unsubscribes, then calls for cleanTomb in a * separate atomic CAS operation. @@ -229,72 +97,17 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { */ @Override public void removeSubscription(Topic topic, String clientID) { - removeFromTree(topic, clientID); + ctrie.removeFromTree(topic, clientID); this.subscriptionsRepository.removeSubscription(topic.toString(), clientID); } - private void removeFromTree(Topic topic, String clientID) { - Action res; - do { - res = remove(clientID, topic, this.root, NO_PARENT); - } while (res == Action.REPEAT); - } - - private Action remove(String clientId, Topic topic, INode inode, INode iParent) { - 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, inode); - } else { - final CNode cnode = inode.mainNode(); - if (cnode instanceof TNode) { - // this inode is a tomb, has no clients and should be cleaned up - // Because we implemented cleanTomb below, this should be rare, but possible - // Consider calling cleanTomb here too - return Action.OK; - } - if (cnode.containsOnly(clientId) && topic.isEmpty() && cnode.allChildren().isEmpty()) { - // last client to leave this node, AND there are no downstream children, remove via TNode tomb - if (inode == this.root) { - return inode.compareAndSet(cnode, inode.mainNode().copy()) ? Action.OK : Action.REPEAT; - } - TNode tnode = new TNode(); - return inode.compareAndSet(cnode, tnode) ? cleanTomb(inode, iParent) : Action.REPEAT; - } else if (cnode.contains(clientId) && topic.isEmpty()) { - CNode updatedCnode = cnode.copy(); - updatedCnode.removeSubscriptionsFor(clientId); - return inode.compareAndSet(cnode, updatedCnode) ? Action.OK : Action.REPEAT; - } else { - //someone else already removed - return Action.OK; - } - } - } - @Override public int size() { - SubscriptionCounterVisitor visitor = new SubscriptionCounterVisitor(); - dfsVisit(this.root, visitor, 0); - return visitor.getResult(); + return ctrie.size(); } @Override 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); - } + return ctrie.dumpTree(); } } diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/DumpTreeVisitor.java b/broker/src/main/java/io/moquette/broker/subscriptions/DumpTreeVisitor.java index bf75d355..c2f6e647 100644 --- a/broker/src/main/java/io/moquette/broker/subscriptions/DumpTreeVisitor.java +++ b/broker/src/main/java/io/moquette/broker/subscriptions/DumpTreeVisitor.java @@ -17,7 +17,7 @@ package io.moquette.broker.subscriptions; import io.netty.util.internal.StringUtil; -class DumpTreeVisitor implements CTrieSubscriptionDirectory.IVisitor { +class DumpTreeVisitor implements CTrie.IVisitor { String s = ""; diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/SubscriptionCounterVisitor.java b/broker/src/main/java/io/moquette/broker/subscriptions/SubscriptionCounterVisitor.java index cd55df1a..2a273aa2 100644 --- a/broker/src/main/java/io/moquette/broker/subscriptions/SubscriptionCounterVisitor.java +++ b/broker/src/main/java/io/moquette/broker/subscriptions/SubscriptionCounterVisitor.java @@ -17,7 +17,7 @@ package io.moquette.broker.subscriptions; import java.util.concurrent.atomic.AtomicInteger; -class SubscriptionCounterVisitor implements CTrieSubscriptionDirectory.IVisitor { +class SubscriptionCounterVisitor implements CTrie.IVisitor { private AtomicInteger accumulator = new AtomicInteger(0); diff --git a/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java b/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java index 4d40a7b8..558efd16 100644 --- a/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java +++ b/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java @@ -25,7 +25,7 @@ import org.junit.Test; import java.util.Optional; import java.util.Set; -import static io.moquette.broker.subscriptions.CTrieSubscriptionDirectoryTest.clientSubOnTopic; +import static io.moquette.broker.subscriptions.CTrieTest.clientSubOnTopic; import static io.moquette.broker.subscriptions.Topic.asTopic; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertEquals; @@ -48,25 +48,25 @@ public class CTrieSubscriptionDirectoryMatchingTest { public void testMatchSimple() { Subscription slashSub = clientSubOnTopic("TempSensor1", "/"); sut.add(slashSub); - assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).isEmpty(); + assertThat(sut.matchWithoutQosSharpening(asTopic("finance"))).isEmpty(); Subscription slashFinanceSub = clientSubOnTopic("TempSensor1", "/finance"); sut.add(slashFinanceSub); - assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).isEmpty(); + assertThat(sut.matchWithoutQosSharpening(asTopic("finance"))).isEmpty(); - assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).contains(slashFinanceSub); - assertThat(sut.recursiveMatch(asTopic("/"), sut.root)).contains(slashSub); + assertThat(sut.matchWithoutQosSharpening(asTopic("/finance"))).contains(slashFinanceSub); + assertThat(sut.matchWithoutQosSharpening(asTopic("/"))).contains(slashSub); } @Test public void testMatchSimpleMulti() { Subscription anySub = clientSubOnTopic("TempSensor1", "#"); sut.add(anySub); - assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).contains(anySub); + assertThat(sut.matchWithoutQosSharpening(asTopic("finance"))).contains(anySub); Subscription financeAnySub = clientSubOnTopic("TempSensor1", "finance/#"); sut.add(financeAnySub); - assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).containsExactlyInAnyOrder(financeAnySub, anySub); + assertThat(sut.matchWithoutQosSharpening(asTopic("finance"))).containsExactlyInAnyOrder(financeAnySub, anySub); } @Test @@ -77,9 +77,9 @@ public class CTrieSubscriptionDirectoryMatchingTest { sut.add(financeAnySub); // Verify - assertThat(sut.recursiveMatch(asTopic("finance/stock"), sut.root)) + assertThat(sut.matchWithoutQosSharpening(asTopic("finance/stock"))) .containsExactlyInAnyOrder(financeAnySub, anySub); - assertThat(sut.recursiveMatch(asTopic("finance/stock/ibm"), sut.root)) + assertThat(sut.matchWithoutQosSharpening(asTopic("finance/stock/ibm"))) .containsExactlyInAnyOrder(financeAnySub, anySub); // System.out.println(sut.dumpTree()); } @@ -90,18 +90,18 @@ public class CTrieSubscriptionDirectoryMatchingTest { sut.add(financeAnySub); // Verify - assertThat(sut.recursiveMatch(asTopic("finance/stock/ibm"), sut.root)).containsExactly(financeAnySub); + assertThat(sut.matchWithoutQosSharpening(asTopic("finance/stock/ibm"))).containsExactly(financeAnySub); } @Test public void testMatchSimpleSingle() { Subscription anySub = clientSubOnTopic("AnySensor", "+"); sut.add(anySub); - assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).containsExactly(anySub); + assertThat(sut.matchWithoutQosSharpening(asTopic("finance"))).containsExactly(anySub); Subscription financeOne = clientSubOnTopic("AnySensor", "finance/+"); sut.add(financeOne); - assertThat(sut.recursiveMatch(asTopic("finance/stock"), sut.root)).containsExactly(financeOne); + assertThat(sut.matchWithoutQosSharpening(asTopic("finance/stock"))).containsExactly(financeOne); } @Test @@ -110,7 +110,7 @@ public class CTrieSubscriptionDirectoryMatchingTest { sut.add(manySub); // verify - assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).contains(manySub); + assertThat(sut.matchWithoutQosSharpening(asTopic("/finance"))).contains(manySub); } @Test @@ -121,8 +121,8 @@ public class CTrieSubscriptionDirectoryMatchingTest { sut.add(anySub); // Verify - assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).containsOnly(slashPlusSub); - assertThat(sut.recursiveMatch(asTopic("/finance"), sut.root)).doesNotContain(anySub); + assertThat(sut.matchWithoutQosSharpening(asTopic("/finance"))).containsOnly(slashPlusSub); + assertThat(sut.matchWithoutQosSharpening(asTopic("/finance"))).doesNotContain(anySub); } @Test @@ -133,7 +133,7 @@ public class CTrieSubscriptionDirectoryMatchingTest { sut.add(slashPlusDeepSub); // Verify - assertThat(sut.recursiveMatch(asTopic("/finance/stock/ibm"), sut.root)) + assertThat(sut.matchWithoutQosSharpening(asTopic("/finance/stock/ibm"))) .containsExactlyInAnyOrder(slashPlusSub, slashPlusDeepSub); } @@ -142,8 +142,8 @@ public class CTrieSubscriptionDirectoryMatchingTest { Subscription sub = clientSubOnTopic("AnySensor1", "#"); sut.add(sub); - assertThat(sut.recursiveMatch(asTopic("finance"), sut.root)).isNotEmpty(); - assertThat(sut.recursiveMatch(asTopic("finance/ibm"), sut.root)).isNotEmpty(); + assertThat(sut.matchWithoutQosSharpening(asTopic("finance"))).isNotEmpty(); + assertThat(sut.matchWithoutQosSharpening(asTopic("finance/ibm"))).isNotEmpty(); } @Test @@ -178,7 +178,7 @@ public class CTrieSubscriptionDirectoryMatchingTest { Subscription sub = clientSubOnTopic("AnySensor1", s); sut.add(sub); - assertThat(sut.recursiveMatch(asTopic(t), sut.root)).isNotEmpty(); + assertThat(sut.matchWithoutQosSharpening(asTopic(t))).isNotEmpty(); } private void assertNotMatch(String subscription, String topic) { @@ -189,7 +189,7 @@ public class CTrieSubscriptionDirectoryMatchingTest { Subscription sub = clientSubOnTopic("AnySensor1", subscription); sut.add(sub); - assertThat(sut.recursiveMatch(asTopic(topic), sut.root)).isEmpty(); + assertThat(sut.matchWithoutQosSharpening(asTopic(topic))).isEmpty(); } @Test @@ -220,7 +220,7 @@ public class CTrieSubscriptionDirectoryMatchingTest { sut.removeSubscription(asTopic("/topic"), slashSub2.clientId); // Verify - Subscription remainedSubscription = sut.recursiveMatch(asTopic("/topic"), sut.root).iterator().next(); + Subscription remainedSubscription = sut.matchWithoutQosSharpening(asTopic("/topic")).iterator().next(); assertThat(remainedSubscription.clientId).isEqualTo(slashSub.clientId); assertEquals(slashSub.clientId, remainedSubscription.clientId); } @@ -234,7 +234,7 @@ public class CTrieSubscriptionDirectoryMatchingTest { sut.removeSubscription(asTopic("/topic"), slashSub.clientId); // Verify - final Set matchingSubscriptions = sut.recursiveMatch(asTopic("/topic"), sut.root); + final Set matchingSubscriptions = sut.matchWithoutQosSharpening(asTopic("/topic")); assertThat(matchingSubscriptions).isEmpty(); } diff --git a/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryTest.java b/broker/src/test/java/io/moquette/broker/subscriptions/CTrieTest.java similarity index 73% rename from broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryTest.java rename to broker/src/test/java/io/moquette/broker/subscriptions/CTrieTest.java index 159b8715..3f15ba09 100644 --- a/broker/src/test/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectoryTest.java +++ b/broker/src/test/java/io/moquette/broker/subscriptions/CTrieTest.java @@ -16,8 +16,6 @@ package io.moquette.broker.subscriptions; -import io.moquette.broker.ISubscriptionsRepository; -import io.moquette.persistence.MemorySubscriptionsRepository; import io.netty.handler.codec.mqtt.MqttQoS; import org.junit.Before; import org.junit.Test; @@ -30,21 +28,19 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; -public class CTrieSubscriptionDirectoryTest { +public class CTrieTest { - private CTrieSubscriptionDirectory sut; + private CTrie sut; @Before public void setUp() { - sut = new CTrieSubscriptionDirectory(); - ISubscriptionsRepository sessionsRepository = new MemorySubscriptionsRepository(); - sut.init(sessionsRepository); + sut = new CTrie(); } @Test public void testAddOnSecondLayerWithEmptyTokenOnEmptyTree() { //Exercise - sut.add(clientSubOnTopic("TempSensor1", "/")); + sut.addToTree(clientSubOnTopic("TempSensor1", "/")); //Verify final Optional matchedNode = sut.lookup(asTopic("/")); @@ -65,7 +61,7 @@ public class CTrieSubscriptionDirectoryTest { @Test public void testAddFirstLayerNodeOnEmptyTree() { //Exercise - sut.add(clientSubOnTopic("TempSensor1", "/temp")); + sut.addToTree(clientSubOnTopic("TempSensor1", "/temp")); //Verify final Optional matchedNode = sut.lookup(asTopic("/temp")); @@ -76,7 +72,7 @@ public class CTrieSubscriptionDirectoryTest { @Test public void testLookup() { final Subscription existingSubscription = clientSubOnTopic("TempSensor1", "/temp"); - sut.add(existingSubscription); + sut.addToTree(existingSubscription); //Exercise final Optional matchedNode = sut.lookup(asTopic("/humidity")); @@ -88,11 +84,11 @@ public class CTrieSubscriptionDirectoryTest { @Test public void testAddNewSubscriptionOnExistingNode() { final Subscription existingSubscription = clientSubOnTopic("TempSensor1", "/temp"); - sut.add(existingSubscription); + sut.addToTree(existingSubscription); //Exercise final Subscription newSubscription = clientSubOnTopic("TempSensor2", "/temp"); - sut.add(newSubscription); + sut.addToTree(newSubscription); //Verify final Optional matchedNode = sut.lookup(asTopic("/temp")); @@ -103,11 +99,11 @@ public class CTrieSubscriptionDirectoryTest { @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")); + sut.addToTree(clientSubOnTopic("TempSensorRM", "/italy/roma/temp")); + sut.addToTree(clientSubOnTopic("TempSensorFI", "/italy/firenze/temp")); + sut.addToTree(clientSubOnTopic("HumSensorFI", "/italy/roma/humidity")); final Subscription happinessSensor = clientSubOnTopic("HappinessSensor", "/italy/happiness"); - sut.add(happinessSensor); + sut.addToTree(happinessSensor); //Verify final Optional matchedNode = sut.lookup(asTopic("/italy/happiness")); @@ -122,10 +118,10 @@ public class CTrieSubscriptionDirectoryTest { @Test public void givenTreeWithSomeNodeWhenRemoveContainedSubscriptionThenNodeIsUpdated() { - sut.add(clientSubOnTopic("TempSensor1", "/temp")); + sut.addToTree(clientSubOnTopic("TempSensor1", "/temp")); //Exercise - sut.removeSubscription(asTopic("/temp"), "TempSensor1"); + sut.removeFromTree(asTopic("/temp"), "TempSensor1"); //Verify final Optional matchedNode = sut.lookup(asTopic("/temp")); @@ -134,22 +130,22 @@ public class CTrieSubscriptionDirectoryTest { @Test public void givenTreeWithSomeNodeUnsubscribeAndResubscribeCleanTomb() { - sut.add(clientSubOnTopic("TempSensor1", "test")); - sut.removeSubscription(asTopic("test"), "TempSensor1"); + sut.addToTree(clientSubOnTopic("TempSensor1", "test")); + sut.removeFromTree(asTopic("test"), "TempSensor1"); - sut.add(clientSubOnTopic("TempSensor1", "test")); + sut.addToTree(clientSubOnTopic("TempSensor1", "test")); assertTrue(sut.root.mainNode().allChildren().size() == 1); // looking to see if TNode is cleaned up } @Test public void givenTreeWithSomeNodeWhenRemoveMultipleTimes() { - sut.add(clientSubOnTopic("TempSensor1", "test")); + sut.addToTree(clientSubOnTopic("TempSensor1", "test")); // make sure no TNode exceptions - sut.removeSubscription(asTopic("test"), "TempSensor1"); - sut.removeSubscription(asTopic("test"), "TempSensor1"); - sut.removeSubscription(asTopic("test"), "TempSensor1"); - sut.removeSubscription(asTopic("test"), "TempSensor1"); + sut.removeFromTree(asTopic("test"), "TempSensor1"); + sut.removeFromTree(asTopic("test"), "TempSensor1"); + sut.removeFromTree(asTopic("test"), "TempSensor1"); + sut.removeFromTree(asTopic("test"), "TempSensor1"); //Verify final Optional matchedNode = sut.lookup(asTopic("/temp")); @@ -158,12 +154,12 @@ public class CTrieSubscriptionDirectoryTest { @Test public void givenTreeWithSomeDeepNodeWhenRemoveMultipleTimes() { - sut.add(clientSubOnTopic("TempSensor1", "/test/me/1/2/3")); + sut.addToTree(clientSubOnTopic("TempSensor1", "/test/me/1/2/3")); // make sure no TNode exceptions - sut.removeSubscription(asTopic("/test/me/1/2/3"), "TempSensor1"); - sut.removeSubscription(asTopic("/test/me/1/2/3"), "TempSensor1"); - sut.removeSubscription(asTopic("/test/me/1/2/3"), "TempSensor1"); + sut.removeFromTree(asTopic("/test/me/1/2/3"), "TempSensor1"); + sut.removeFromTree(asTopic("/test/me/1/2/3"), "TempSensor1"); + sut.removeFromTree(asTopic("/test/me/1/2/3"), "TempSensor1"); //Verify final Optional matchedNode = sut.lookup(asTopic("/temp")); @@ -172,14 +168,14 @@ public class CTrieSubscriptionDirectoryTest { @Test public void givenTreeWithSomeNodeHierarchWhenRemoveContainedSubscriptionThenNodeIsUpdated() { - sut.add(clientSubOnTopic("TempSensor1", "/temp/1")); - sut.add(clientSubOnTopic("TempSensor1", "/temp/2")); + sut.addToTree(clientSubOnTopic("TempSensor1", "/temp/1")); + sut.addToTree(clientSubOnTopic("TempSensor1", "/temp/2")); //Exercise - sut.removeSubscription(asTopic("/temp/1"), "TempSensor1"); + sut.removeFromTree(asTopic("/temp/1"), "TempSensor1"); - sut.removeSubscription(asTopic("/temp/1"), "TempSensor1"); - final Set matchingSubs = sut.recursiveMatch(asTopic("/temp/2"), sut.root); + sut.removeFromTree(asTopic("/temp/1"), "TempSensor1"); + final Set matchingSubs = sut.recursiveMatch(asTopic("/temp/2")); //Verify final Subscription expectedMatchingsub = new Subscription("TempSensor1", asTopic("/temp/2"), MqttQoS.AT_MOST_ONCE); @@ -188,14 +184,14 @@ public class CTrieSubscriptionDirectoryTest { @Test public void givenTreeWithSomeNodeHierarchWhenRemoveContainedSubscriptionSmallerThenNodeIsNotUpdated() { - sut.add(clientSubOnTopic("TempSensor1", "/temp/1")); - sut.add(clientSubOnTopic("TempSensor1", "/temp/2")); + sut.addToTree(clientSubOnTopic("TempSensor1", "/temp/1")); + sut.addToTree(clientSubOnTopic("TempSensor1", "/temp/2")); //Exercise - sut.removeSubscription(asTopic("/temp"), "TempSensor1"); + sut.removeFromTree(asTopic("/temp"), "TempSensor1"); - final Set matchingSubs1 = sut.recursiveMatch(asTopic("/temp/1"), sut.root); - final Set matchingSubs2 = sut.recursiveMatch(asTopic("/temp/2"), sut.root); + final Set matchingSubs1 = sut.recursiveMatch(asTopic("/temp/1")); + final Set matchingSubs2 = sut.recursiveMatch(asTopic("/temp/2")); //Verify // not clear to me, but I believe /temp unsubscribe should not unsub you from downstream /temp/1 or /temp/2 @@ -207,9 +203,9 @@ public class CTrieSubscriptionDirectoryTest { @Test public void givenTreeWithDeepNodeWhenRemoveContainedSubscriptionThenNodeIsUpdated() { - sut.add(clientSubOnTopic("TempSensor1", "/bah/bin/bash")); + sut.addToTree(clientSubOnTopic("TempSensor1", "/bah/bin/bash")); - sut.removeSubscription(asTopic("/bah/bin/bash"), "TempSensor1"); + sut.removeFromTree(asTopic("/bah/bin/bash"), "TempSensor1"); //Verify final Optional matchedNode = sut.lookup(asTopic("/bah/bin/bash")); @@ -218,10 +214,10 @@ public class CTrieSubscriptionDirectoryTest { @Test public void testMatchSubscriptionNoWildcards() { - sut.add(clientSubOnTopic("TempSensor1", "/temp")); + sut.addToTree(clientSubOnTopic("TempSensor1", "/temp")); //Exercise - final Set matchingSubs = sut.recursiveMatch(asTopic("/temp"), sut.root); + final Set matchingSubs = sut.recursiveMatch(asTopic("/temp")); //Verify final Subscription expectedMatchingsub = new Subscription("TempSensor1", asTopic("/temp"), MqttQoS.AT_MOST_ONCE); @@ -230,12 +226,12 @@ public class CTrieSubscriptionDirectoryTest { @Test public void testRemovalInnerTopicOffRootSameClient() { - sut.add(clientSubOnTopic("TempSensor1", "temp")); - sut.add(clientSubOnTopic("TempSensor1", "temp/1")); + sut.addToTree(clientSubOnTopic("TempSensor1", "temp")); + sut.addToTree(clientSubOnTopic("TempSensor1", "temp/1")); //Exercise - final Set matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root); - final Set matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + final Set matchingSubs1 = sut.recursiveMatch(asTopic("temp")); + final Set matchingSubs2 = sut.recursiveMatch(asTopic("temp/1")); //Verify final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp"), MqttQoS.AT_MOST_ONCE); @@ -244,11 +240,11 @@ public class CTrieSubscriptionDirectoryTest { assertThat(matchingSubs1).contains(expectedMatchingsub1); assertThat(matchingSubs2).contains(expectedMatchingsub2); - sut.removeSubscription(asTopic("temp"), "TempSensor1"); + sut.removeFromTree(asTopic("temp"), "TempSensor1"); //Exercise - final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); - final Set matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp")); + final Set matchingSubs4 = sut.recursiveMatch(asTopic("temp/1")); assertThat(matchingSubs3).doesNotContain(expectedMatchingsub1); assertThat(matchingSubs4).contains(expectedMatchingsub2); @@ -256,12 +252,12 @@ public class CTrieSubscriptionDirectoryTest { @Test public void testRemovalInnerTopicOffRootDiffClient() { - sut.add(clientSubOnTopic("TempSensor1", "temp")); - sut.add(clientSubOnTopic("TempSensor2", "temp/1")); + sut.addToTree(clientSubOnTopic("TempSensor1", "temp")); + sut.addToTree(clientSubOnTopic("TempSensor2", "temp/1")); //Exercise - final Set matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root); - final Set matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + final Set matchingSubs1 = sut.recursiveMatch(asTopic("temp")); + final Set matchingSubs2 = sut.recursiveMatch(asTopic("temp/1")); //Verify final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp"), MqttQoS.AT_MOST_ONCE); @@ -270,11 +266,11 @@ public class CTrieSubscriptionDirectoryTest { assertThat(matchingSubs1).contains(expectedMatchingsub1); assertThat(matchingSubs2).contains(expectedMatchingsub2); - sut.removeSubscription(asTopic("temp"), "TempSensor1"); + sut.removeFromTree(asTopic("temp"), "TempSensor1"); //Exercise - final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); - final Set matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp")); + final Set matchingSubs4 = sut.recursiveMatch(asTopic("temp/1")); assertThat(matchingSubs3).doesNotContain(expectedMatchingsub1); assertThat(matchingSubs4).contains(expectedMatchingsub2); @@ -282,12 +278,12 @@ public class CTrieSubscriptionDirectoryTest { @Test public void testRemovalOuterTopicOffRootDiffClient() { - sut.add(clientSubOnTopic("TempSensor1", "temp")); - sut.add(clientSubOnTopic("TempSensor2", "temp/1")); + sut.addToTree(clientSubOnTopic("TempSensor1", "temp")); + sut.addToTree(clientSubOnTopic("TempSensor2", "temp/1")); //Exercise - final Set matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root); - final Set matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + final Set matchingSubs1 = sut.recursiveMatch(asTopic("temp")); + final Set matchingSubs2 = sut.recursiveMatch(asTopic("temp/1")); //Verify final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp"), MqttQoS.AT_MOST_ONCE); @@ -296,11 +292,11 @@ public class CTrieSubscriptionDirectoryTest { assertThat(matchingSubs1).contains(expectedMatchingsub1); assertThat(matchingSubs2).contains(expectedMatchingsub2); - sut.removeSubscription(asTopic("temp/1"), "TempSensor2"); + sut.removeFromTree(asTopic("temp/1"), "TempSensor2"); //Exercise - final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); - final Set matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp")); + final Set matchingSubs4 = sut.recursiveMatch(asTopic("temp/1")); assertThat(matchingSubs3).contains(expectedMatchingsub1); assertThat(matchingSubs4).doesNotContain(expectedMatchingsub2);