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 index 74a29df0..4c722dcc 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CNode.java @@ -24,9 +24,7 @@ class CNode { Token token; private List children; - Set subscriptions = new HashSet<>(); - - //private int subtreeSubscriptions; + Set subscriptions; CNode() { this.children = new ArrayList<>(); @@ -35,9 +33,9 @@ class CNode { //Copy constructor private CNode(Token token, List children, Set subscriptions) { - this.token = token; - this.children = children; - this.subscriptions = subscriptions; + this.token = token; // keep reference, root comparison in directory logic relies on it for now. + this.subscriptions = new HashSet<>(subscriptions); + this.children = new ArrayList<>(children); } boolean anyChildrenMatch(Token token) { @@ -75,6 +73,9 @@ class CNode { public void add(INode newINode) { this.children.add(newINode); } + public void remove(INode node) { + this.children.remove(node); + } CNode addSubscription(String clientId, Topic topic) { this.subscriptions.add(new Subscription(clientId, topic)); @@ -83,6 +84,7 @@ class CNode { /** * @return true iff the subscriptions contained in this node are owned by clientId + * AND at least one subscription is actually present for that clientId * */ boolean containsOnly(String clientId) { for (Subscription sub : this.subscriptions) { @@ -90,7 +92,7 @@ class CNode { return false; } } - return true; + return !this.subscriptions.isEmpty(); } //TODO this is equivalent to negate(containsOnly(clientId)) 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 index 56968c61..47e3b77f 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectory.java @@ -31,6 +31,7 @@ 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 volatile SessionsRepository sessionsRepository; @@ -49,10 +50,13 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { @Override public void visit(CNode node, int deep) { String indentTabs = indentTabs(deep); - s += indentTabs + (node.token == null ? "" : node.token.toString()) + prettySubscriptions(node) + "\n"; + s += indentTabs + (node.token == null ? "''" : node.token.toString()) + prettySubscriptions(node) + "\n"; } private String prettySubscriptions(CNode node) { + if (node instanceof TNode) { + return "TNode"; + } if (node.subscriptions.isEmpty()) { return StringUtil.EMPTY_STRING; } @@ -73,6 +77,7 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { private String indentTabs(int deep) { StringBuilder s = new StringBuilder(); if (deep > 0) { + s.append(" "); for (int i = 0; i < deep - 1; i++) { s.append("| "); } @@ -196,6 +201,23 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { 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 + * @param iParent + * @return + */ + 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; + } public void add(Subscription newSubscription) { Action res; @@ -259,25 +281,44 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { } } + /** + * + * Removes subscription from CTrie, adds TNode when the last client unsubscribes, then calls for cleanTomb in a + * seperate atomic CAS operation. + * + * + * @param topic + * @param clientID + */ public void removeSubscription(Topic topic, String clientID) { Action res; do { - res = remove(clientID, topic, this.root); + res = remove(clientID, topic, this.root, NO_PARENT); } while (res == Action.REPEAT); } - private Action remove(String clientId, Topic topic, INode inode) { + 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); + return remove(clientId, remainingTopic, nextInode, inode); } else { final CNode cnode = inode.mainNode(); - if (cnode.containsOnly(clientId)) { + 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) ? Action.OK : Action.REPEAT; - } else if (cnode.contains(clientId)) { + 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; 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 index 0dbeec59..2385d77c 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/INode.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/INode.java @@ -22,6 +22,9 @@ class INode { INode(CNode mainNode) { this.mainNode.set(mainNode); + if (mainNode instanceof TNode) { // this should never happen + throw new IllegalStateException("TNode should not be set on mainNnode"); + } } boolean compareAndSet(CNode old, CNode newNode) { @@ -35,4 +38,8 @@ class INode { CNode mainNode() { return this.mainNode.get(); } + + boolean isTombed() { + return this.mainNode() instanceof TNode; + } } 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 index 9fb34588..9ce84dc4 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java @@ -132,6 +132,95 @@ public class CTrieSubscriptionDirectoryTest { assertFalse("Node on path /temp can't be present", matchedNode.isPresent()); } + @Test + public void givenTreeWithSomeNodeUnsubscribeAndResubscribeCleanTomb() { + sut.add(clientSubOnTopic("TempSensor1", "test")); + sut.removeSubscription(asTopic("test"), "TempSensor1"); + + sut.add(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")); + + // 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"); + + //Verify + final Optional matchedNode = sut.lookup(asTopic("/temp")); + assertFalse("Node on path /temp can't be present", matchedNode.isPresent()); + } + + @Test + public void givenTreeWithSomeDeepNodeWhenRemoveMultipleTimes() { + sut.add(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"); + + //Verify + final Optional matchedNode = sut.lookup(asTopic("/temp")); + assertFalse("Node on path /temp can't be present", matchedNode.isPresent()); + } + + @Test + public void givenTreeWithSomeNodeHierarchWhenRemoveContainedSubscriptionThenNodeIsUpdated() { + sut.add(clientSubOnTopic("TempSensor1", "/temp/1")); + sut.add(clientSubOnTopic("TempSensor1", "/temp/2")); + + //Exercise + sut.removeSubscription(asTopic("/temp/1"), "TempSensor1"); + + sut.removeSubscription(asTopic("/temp/1"), "TempSensor1"); + final Set matchingSubs = sut.recursiveMatch(asTopic("/temp/2"), sut.root); + + //Verify + final Subscription expectedMatchingsub = new Subscription("TempSensor1", asTopic("/temp/2")); + assertThat(matchingSubs).contains(expectedMatchingsub); + } + + @Test + public void givenTreeWithSomeNodeHierarchWhenRemoveContainedSubscriptionSmallerThenNodeIsNotUpdated() { + sut.add(clientSubOnTopic("TempSensor1", "/temp/1")); + sut.add(clientSubOnTopic("TempSensor1", "/temp/2")); + + //Exercise + + sut.removeSubscription(asTopic("/temp"), "TempSensor1"); + + final Set matchingSubs1 = sut.recursiveMatch(asTopic("/temp/1"), sut.root); + final Set matchingSubs2 = sut.recursiveMatch(asTopic("/temp/2"), sut.root); + + //Verify + + // not clear to me, but I believe /temp unsubscribe should not unsub you from downstream /temp/1 or /temp/2 + + final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("/temp/1")); + assertThat(matchingSubs1).contains(expectedMatchingsub1); + final Subscription expectedMatchingsub2 = new Subscription("TempSensor1", asTopic("/temp/2")); + assertThat(matchingSubs2).contains(expectedMatchingsub2); + } + + + @Test + public void givenTreeWithDeepNodeWhenRemoveContainedSubscriptionThenNodeIsUpdated() { + sut.add(clientSubOnTopic("TempSensor1", "/bah/bin/bash")); + + sut.removeSubscription(asTopic("/bah/bin/bash"),"TempSensor1"); + + //Verify + final Optional matchedNode = sut.lookup(asTopic("/bah/bin/bash")); + assertFalse("Node on path /temp can't be present", matchedNode.isPresent()); + } + + @Test public void testMatchSubscriptionNoWildcards() { sut.add(clientSubOnTopic("TempSensor1", "/temp")); @@ -143,4 +232,84 @@ public class CTrieSubscriptionDirectoryTest { final Subscription expectedMatchingsub = new Subscription("TempSensor1", asTopic("/temp")); assertThat(matchingSubs).contains(expectedMatchingsub); } + + @Test + public void testRemovalInnerTopicOffRootSameClient() { + sut.add(clientSubOnTopic("TempSensor1", "temp")); + sut.add(clientSubOnTopic("TempSensor1", "temp/1")); + + //Exercise + final Set matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + + //Verify + final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp")); + final Subscription expectedMatchingsub2 = new Subscription("TempSensor1", asTopic("temp/1")); + + assertThat(matchingSubs1).contains(expectedMatchingsub1); + assertThat(matchingSubs2).contains(expectedMatchingsub2); + + sut.removeSubscription(asTopic("temp"),"TempSensor1"); + + //Exercise + final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + + assertThat(matchingSubs3).doesNotContain(expectedMatchingsub1); + assertThat(matchingSubs4).contains(expectedMatchingsub2); + } + + @Test + public void testRemovalInnerTopicOffRootDiffClient() { + sut.add(clientSubOnTopic("TempSensor1", "temp")); + sut.add(clientSubOnTopic("TempSensor2", "temp/1")); + + //Exercise + final Set matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + + //Verify + final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp")); + final Subscription expectedMatchingsub2 = new Subscription("TempSensor2", asTopic("temp/1")); + + assertThat(matchingSubs1).contains(expectedMatchingsub1); + assertThat(matchingSubs2).contains(expectedMatchingsub2); + + sut.removeSubscription(asTopic("temp"),"TempSensor1"); + + //Exercise + final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + + assertThat(matchingSubs3).doesNotContain(expectedMatchingsub1); + assertThat(matchingSubs4).contains(expectedMatchingsub2); + } + + @Test + public void testRemovalOuterTopicOffRootDiffClient() { + sut.add(clientSubOnTopic("TempSensor1", "temp")); + sut.add(clientSubOnTopic("TempSensor2", "temp/1")); + + //Exercise + final Set matchingSubs1 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set matchingSubs2 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + + //Verify + final Subscription expectedMatchingsub1 = new Subscription("TempSensor1", asTopic("temp")); + final Subscription expectedMatchingsub2 = new Subscription("TempSensor2", asTopic("temp/1")); + + assertThat(matchingSubs1).contains(expectedMatchingsub1); + assertThat(matchingSubs2).contains(expectedMatchingsub2); + + sut.removeSubscription(asTopic("temp/1"),"TempSensor2"); + + //Exercise + final Set matchingSubs3 = sut.recursiveMatch(asTopic("temp"), sut.root); + final Set matchingSubs4 = sut.recursiveMatch(asTopic("temp/1"), sut.root); + + assertThat(matchingSubs3).contains(expectedMatchingsub1); + assertThat(matchingSubs4).doesNotContain(expectedMatchingsub2); + } } + +