From f45c0a76e1f47b1e91295bda13c1de08d04ea352 Mon Sep 17 00:00:00 2001 From: andsel Date: Mon, 28 Sep 2015 23:40:28 +0200 Subject: [PATCH] Changed the subscription tree to work as persistent tree --- .../moquette/spi/impl/ProtocolProcessor.java | 5 +- .../subscriptions/SubscriptionsStore.java | 92 ++++++++++++++----- .../spi/impl/subscriptions/Token.java | 2 +- .../spi/impl/subscriptions/TreeNode.java | 22 ++++- .../subscriptions/SubscriptionsStoreTest.java | 28 +++++- 5 files changed, 121 insertions(+), 28 deletions(-) diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 36dc94f4..ee130aac 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -20,6 +20,7 @@ import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.dsl.Disruptor; import java.nio.ByteBuffer; import java.util.*; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -96,7 +97,7 @@ public class ProtocolProcessor implements EventHandler { private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); - private Map m_clientIDs = new HashMap<>(); + private Map m_clientIDs = new ConcurrentHashMap<>(); private SubscriptionsStore subscriptions; private boolean allowAnonymous; private IAuthorizator m_authorizator; @@ -106,7 +107,7 @@ public class ProtocolProcessor implements EventHandler { private BrokerInterceptor m_interceptor; //maps clientID to Will testament, if specified on CONNECT - private Map m_willStore = new HashMap<>(); + private Map m_willStore = new ConcurrentHashMap<>(); private ExecutorService m_executor; private RingBuffer m_ringBuffer; diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java index 84adac23..404bd289 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -18,6 +18,7 @@ package org.eclipse.moquette.spi.impl.subscriptions; import java.text.ParseException; import java.util.*; import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.atomic.AtomicReference; import org.eclipse.moquette.spi.ISessionsStore; import org.slf4j.Logger; @@ -30,6 +31,16 @@ import org.slf4j.LoggerFactory; */ public class SubscriptionsStore { + public static class NodeCouple { + final TreeNode root; + final TreeNode createdNode; + + public NodeCouple(TreeNode root, TreeNode createdNode) { + this.root = root; + this.createdNode = createdNode; + } + } + /** * Check if the topic filter of the subscription is well formed * */ @@ -43,7 +54,7 @@ public class SubscriptionsStore { } } - public static interface IVisitor { + public interface IVisitor { void visit(TreeNode node, int deep); T getResult(); @@ -90,7 +101,7 @@ public class SubscriptionsStore { } } - private TreeNode subscriptions = new TreeNode(null); + private AtomicReference subscriptions = new AtomicReference<>(new TreeNode(null)); private ISessionsStore m_sessionsStore; private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class); @@ -115,12 +126,19 @@ public class SubscriptionsStore { LOG.debug("Finished loading. Subscription tree after {}", dumpTree()); } } - + + //TODO modify the data structure protected void addDirect(Subscription newSubscription) { - TreeNode current = findMatchingNode(newSubscription.topicFilter); - current.addSubscription(newSubscription); + TreeNode oldRoot; + NodeCouple couple; + do { + oldRoot = subscriptions.get(); + couple = recreatePath(newSubscription.topicFilter, oldRoot); + couple.createdNode.addSubscription(newSubscription); //createdNode could be null? + //spin lock repeating till we can, swap root, if can't swap just re-do the operation + } while(!subscriptions.compareAndSet(oldRoot, couple.root)); } - + private TreeNode findMatchingNode(String topic) { List tokens = new ArrayList<>(); try { @@ -131,13 +149,44 @@ public class SubscriptionsStore { // return; } - TreeNode current = subscriptions; + TreeNode current = subscriptions.get(); for (Token token : tokens) { TreeNode matchingChildren; //check if a children with the same token already exists if ((matchingChildren = current.childWithToken(token)) != null) { current = matchingChildren; + } else { + //can't find any match + return null; + } + } + return current; + } + + protected NodeCouple recreatePath(String topic, final TreeNode oldRoot) { + List tokens = new ArrayList<>(); + try { + tokens = parseTopic(topic); + } catch (ParseException ex) { + //TODO handle the parse exception + LOG.error(null, ex); + } + + final TreeNode newRoot = oldRoot.copy(); + TreeNode parent = newRoot; + TreeNode current = newRoot; + for (Token token : tokens) { + TreeNode matchingChildren; + + //check if a children with the same token already exists + if ((matchingChildren = current.childWithToken(token)) != null) { + //copy the traversed node + current = matchingChildren.copy(); + current.m_parent = parent; + //update the child just added in the children list + parent.updateChild(matchingChildren, current); + parent = current; } else { //create a new node for the newly inserted token matchingChildren = new TreeNode(current); @@ -146,15 +195,11 @@ public class SubscriptionsStore { current = matchingChildren; } } - return current; + return new NodeCouple(newRoot, current); } public void add(Subscription newSubscription) { addDirect(newSubscription); - - //log the subscription -// String clientID = newSubscription.getClientId(); -// m_storageService.addNewSubscription(newSubscription, clientID); } @@ -176,15 +221,15 @@ public class SubscriptionsStore { } /** - * TODO implement testing * It's public because needed in tests :-( bleah */ public void clearAllSubscriptions() { SubscriptionTreeCollector subsCollector = new SubscriptionTreeCollector(); - bfsVisit(subscriptions, subsCollector, 0); + bfsVisit(subscriptions.get(), subsCollector, 0); List allSubscriptions = subsCollector.getResult(); for (Subscription subscription : allSubscriptions) { + //TODO modify the data structure removeSubscription(subscription.getTopicFilter(), subscription.getClientId()); } } @@ -192,24 +237,27 @@ public class SubscriptionsStore { /** * Visit the topics tree to remove matching subscriptions with clientID */ + //TODO modify the data structure public void removeForClient(String clientID) { - subscriptions.removeClientSubscriptions(clientID); + subscriptions.get().removeClientSubscriptions(clientID); //persist the update m_sessionsStore.wipeSubscriptions(clientID); } + //TODO modify the data structure public void deactivate(String clientID) { - subscriptions.deactivate(clientID); + subscriptions.get().deactivate(clientID); //persist the update - Set subs = subscriptions.findAllByClientID(clientID); + Set subs = subscriptions.get().findAllByClientID(clientID); m_sessionsStore.updateSubscriptions(clientID, subs); } + //TODO modify the data structure public void activate(String clientID) { LOG.debug("Activating subscriptions for clientID <{}>", clientID); - subscriptions.activate(clientID); + subscriptions.get().activate(clientID); //persist the update - Set subs = subscriptions.findAllByClientID(clientID); + Set subs = subscriptions.get().findAllByClientID(clientID); m_sessionsStore.updateSubscriptions(clientID, subs); } @@ -230,7 +278,7 @@ public class SubscriptionsStore { Queue tokenQueue = new LinkedBlockingDeque<>(tokens); List matchingSubs = new ArrayList<>(); - subscriptions.matches(tokenQueue, matchingSubs); + subscriptions.get().matches(tokenQueue, matchingSubs); //remove the overlapping subscriptions, selecting ones with greatest qos Map subsForClient = new HashMap<>(); @@ -249,12 +297,12 @@ public class SubscriptionsStore { } public int size() { - return subscriptions.size(); + return subscriptions.get().size(); } public String dumpTree() { DumpTreeVisitor visitor = new DumpTreeVisitor(); - bfsVisit(subscriptions, visitor, 0); + bfsVisit(subscriptions.get(), visitor, 0); return visitor.getResult(); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Token.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Token.java index 3c4d373c..9885f3ca 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Token.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/Token.java @@ -23,7 +23,7 @@ public class Token { static final Token EMPTY = new Token(""); static final Token MULTI = new Token("#"); static final Token SINGLE = new Token("+"); - String name; + final String name; protected Token(String s) { name = s; diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java index 11fd6928..58dca1ed 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java @@ -64,9 +64,22 @@ class TreeNode { void addChild(TreeNode child) { m_children.add(child); } +// +// boolean isLeaf() { +// return m_children.isEmpty(); +// } - boolean isLeaf() { - return m_children.isEmpty(); + /** + * Creates a shallow copy of the current node. + * Copy the token and the children. + * */ + TreeNode copy() { + final TreeNode copy = new TreeNode(this); + copy.m_parent = m_parent; + copy.m_children = new ArrayList<>(m_children); + copy.m_subscriptions = new ArrayList<>(m_subscriptions); + copy.m_token = m_token; + return copy; } /** @@ -83,6 +96,11 @@ class TreeNode { return null; } + void updateChild(TreeNode oldChild, TreeNode newChild) { + m_children.remove(oldChild); + m_children.add(newChild); + } + List subscriptions() { return m_subscriptions; } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index 2e83575f..be5295c5 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -76,7 +76,7 @@ public class SubscriptionsStoreTest { } @Test(expected = ParseException.class) - public void testParseTopicMultiNotAferSeparatorNotValid() throws ParseException { + public void testParseTopicMultiNotAfterSeparatorNotValid() throws ParseException { store.parseTopic("finance#"); } @@ -397,6 +397,32 @@ public class SubscriptionsStoreTest { assertFalse(subscriptions.contains(client1SubQoS0)); //client1SubQoS2 should override client1SubQoS0 } + @Test + public void testRecreatePath_emptyRoot() { + TreeNode oldRoot = new TreeNode(null); + final SubscriptionsStore.NodeCouple resp = store.recreatePath("/finance", oldRoot); + + //Verify + assertNotNull(resp.root); + assertNull(resp.root.m_token); + assertEquals(1, resp.root.m_children.size()); + assertEquals(resp.createdNode, resp.root.m_children.get(0).m_children.get(0)); + } + + @Test + public void testRecreatePath_1layer_tree() { + TreeNode oldRoot = new TreeNode(null); + final SubscriptionsStore.NodeCouple respFinance = store.recreatePath("/finance", oldRoot); + final SubscriptionsStore.NodeCouple respPlus = store.recreatePath("/+", respFinance.root); + + //Verify + assertNotNull(respPlus.root); + assertNull(respPlus.root.m_token); + assertEquals(1, respPlus.root.m_children.size()); + assertTrue(respPlus.root.m_children.get(0).m_children.contains(respPlus.createdNode)); + assertTrue(respPlus.root.m_children.get(0).m_children.contains(respFinance.createdNode)); + } + private static Token[] asArray(Object... l) { Token[] tokens = new Token[l.length]; for (int i = 0; i < l.length; i++) {