From 1173406b0a9c8b6e9271ea882ecbc15f3f5c8c73 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Wed, 8 Aug 2012 21:52:18 +0200 Subject: [PATCH] Fixed matching multi topic and single selector --- .../spi/impl/SubscriptionsStore.java | 80 ++++------ .../spi/impl/SubscriptionsStoreTest.java | 141 ++++++++++++++---- 2 files changed, 143 insertions(+), 78 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java index 2a041da7..d9a5f219 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java @@ -4,6 +4,7 @@ import java.text.ParseException; import java.util.ArrayList; import java.util.List; import java.util.Queue; +import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.LinkedBlockingQueue; import java.util.logging.Level; import java.util.logging.Logger; @@ -71,39 +72,35 @@ public class SubscriptionsStore { void matches(Queue tokens, List matchingSubs) { Token t = tokens.poll(); + + //check if t is null <=> tokens finished if (t == null) { matchingSubs.addAll(m_subscriptions); + //check if it has got a MULTI child and add its subscriptions + for (TreeNode n : m_children) { + if (n.getToken() == Token.MULTI) { + matchingSubs.addAll(n.subscriptions()); + } + } + return; } - //check the matching of the current token - //TODO take care of MULTI and SINGLE -// if (m_token == Token.MULTI) { -// matchingSubs.addAll(m_subscriptions); -// return; -// } - - List matchingChildren = new ArrayList(); - for (TreeNode child : m_children) { - if (child.getToken().match(t)) { - matchingChildren.add(child); - } + //we are on MULTI, than add subscriptions and return + if (m_token == Token.MULTI) { + matchingSubs.addAll(m_subscriptions); + return; } -// if (matchingChildren.isEmpty()) { -// return; -// } - - for (TreeNode child : matchingChildren) { - if (child.getToken() == Token.MULTI) { - matchingSubs.addAll(m_subscriptions); - } else { - child.matches(tokens, matchingSubs); + for (TreeNode n : m_children) { + if (n.getToken().match(t)) { + //Create a copy of token, alse if navigate 2 sibling it + //consumes 2 elements on the queue instead of one + n.matches(new LinkedBlockingQueue(tokens), matchingSubs); } } } - /** * Return the number of registered subscriptions */ @@ -132,14 +129,6 @@ public class SubscriptionsStore { return name; } - /** - * Return true iff the current token is compatible with t. - * - * TODO define the rules of compatibility - * if this is a MULTI or SINGLE and the other is any string => match - * of this is a normal token and also the other, and they are equals => match - * if the other is not normal token => don't match - */ protected boolean match(Token t) { if (t == MULTI || t == SINGLE) { return false; @@ -147,9 +136,9 @@ public class SubscriptionsStore { if (this == MULTI || this == SINGLE) { return true; - } else { - return equals(t); } + + return equals(t); } @Override @@ -181,7 +170,7 @@ public class SubscriptionsStore { } // private List subscriptions = new ArrayList(); - private TreeNode subscriptionsRoot = new TreeNode(null); + private TreeNode subscriptions = new TreeNode(null); public void add(Subscription newSubscription) { List tokens = new ArrayList(); @@ -192,8 +181,7 @@ public class SubscriptionsStore { Logger.getLogger(SubscriptionsStore.class.getName()).log(Level.SEVERE, null, ex); } - //Deep into the tree creating missing nodes where necessary - TreeNode current = subscriptionsRoot; + TreeNode current = subscriptions; for (Token token : tokens) { TreeNode matchingChildren; @@ -219,21 +207,11 @@ public class SubscriptionsStore { //TODO handle the parse exception Logger.getLogger(SubscriptionsStore.class.getName()).log(Level.SEVERE, null, ex); } -// TreeNode current = subscriptionsRoot; - List matchingSubscriptions = new ArrayList(); - subscriptionsRoot.matches(new LinkedBlockingQueue(tokens), matchingSubscriptions); - return matchingSubscriptions; -// for (Token token : tokens) { -// current = current.matchNext(token); -// if (current == null) { -// break; -// } -// } -// if (current != null) { -// return current.subscriptions(); -// } else { -// return Collections.EMPTY_LIST; -// } + + Queue tokenQueue = new LinkedBlockingDeque(tokens); + List matchingSubs = new ArrayList(); + subscriptions.matches(tokenQueue, matchingSubs); + return matchingSubs; } public boolean contains(Subscription sub) { @@ -241,7 +219,7 @@ public class SubscriptionsStore { } public int size() { - return subscriptionsRoot.size(); + return subscriptions.size(); } protected List splitTopic(String topic) throws ParseException { diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java index 5bcbf0bd..d45670c6 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java @@ -14,12 +14,12 @@ import org.junit.Test; * @author andrea */ public class SubscriptionsStoreTest { - + private SubscriptionsStore store; - + public SubscriptionsStoreTest() { } - + @Before public void setUp() { store = new SubscriptionsStore(); @@ -29,91 +29,178 @@ public class SubscriptionsStoreTest { public void testSplitTopic() throws ParseException { List tokens = store.splitTopic("finance/stock/ibm"); assertEqualsSeq(asArray("finance", "stock", "ibm"), tokens); - + tokens = store.splitTopic("/finance/stock/ibm"); assertEqualsSeq(asArray(Token.EMPTY, "finance", "stock", "ibm"), tokens); - + tokens = store.splitTopic("/"); assertEqualsSeq(asArray(Token.EMPTY), tokens); } - - + @Test(expected = ParseException.class) public void testSplitTopicTwinsSlashAvoided() throws ParseException { store.splitTopic("/finance//stock/ibm"); } - + @Test public void testSplitTopicMultiValid() throws ParseException { List tokens = store.splitTopic("finance/stock/#"); assertEqualsSeq(asArray("finance", "stock", Token.MULTI), tokens); - + tokens = store.splitTopic("#"); assertEqualsSeq(asArray(Token.MULTI), tokens); } - - + @Test(expected = ParseException.class) public void testSplitTopicMultiInTheMiddleNotValid() throws ParseException { store.splitTopic("finance/#/closingprice"); } - + @Test(expected = ParseException.class) public void testSplitTopicMultiNotAferSeparatorNotValid() throws ParseException { store.splitTopic("finance#"); } - - + @Test public void testSplitTopicSingleValid() throws ParseException { List tokens = store.splitTopic("finance/stock/+"); assertEqualsSeq(asArray("finance", "stock", Token.SINGLE), tokens); - + tokens = store.splitTopic("+"); assertEqualsSeq(asArray(Token.SINGLE), tokens); - + tokens = store.splitTopic("finance/+/ibm"); assertEqualsSeq(asArray("finance", Token.SINGLE, "ibm"), tokens); } - + @Test(expected = ParseException.class) public void testSplitTopicSingleNotAferSeparatorNotValid() throws ParseException { store.splitTopic("finance+"); } - + @Test public void testMatchSimple() { - store.add(new Subscription(null, "/", AbstractMessage.QOSType.MOST_ONE)); + Subscription slashSub = new Subscription(null, "/", AbstractMessage.QOSType.MOST_ONE); + store.add(slashSub); assertTrue(store.matches("finance").isEmpty()); - store.add(new Subscription(null, "/finance", AbstractMessage.QOSType.MOST_ONE)); + Subscription slashFinanceSub = new Subscription(null, "/finance", AbstractMessage.QOSType.MOST_ONE); + store.add(slashFinanceSub); assertTrue(store.matches("finance").isEmpty()); + + assertTrue(store.matches("/finance").contains(slashFinanceSub)); + assertTrue(store.matches("/").contains(slashSub)); } @Test public void testMatchSimpleMulti() { - store.add(new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE)); - assertFalse(store.matches("finance").isEmpty()); + Subscription anySub = new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE); + store.add(anySub); + assertTrue(store.matches("finance").contains(anySub)); - store.add(new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE)); - assertFalse(store.matches("finance").isEmpty()); + Subscription financeAnySub = new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE); + store.add(financeAnySub); + assertTrue(store.matches("finance").containsAll(Arrays.asList(financeAnySub, anySub))); + } + + @Test + public void testMatchingDeepMulti_one_layer() { + Subscription anySub = new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE); + Subscription financeAnySub = new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE); + store.add(anySub); + store.add(financeAnySub); + + //Verify + assertTrue(store.matches("finance/stock").containsAll(Arrays.asList(financeAnySub, anySub))); + assertTrue(store.matches("finance/stock/ibm").containsAll(Arrays.asList(financeAnySub, anySub))); } + @Test + public void testMatchingDeepMulti_two_layer() { + Subscription financeAnySub = new Subscription(null, "finance/stock/#", AbstractMessage.QOSType.MOST_ONE); + store.add(financeAnySub); + + //Verify + assertTrue(store.matches("finance/stock/ibm").contains(financeAnySub)); + } + + @Test + public void testMatchSimpleSingle() { + Subscription anySub = new Subscription(null, "+", AbstractMessage.QOSType.MOST_ONE); + store.add(anySub); + assertTrue(store.matches("finance").contains(anySub)); + + Subscription financeOne = new Subscription(null, "finance/+", AbstractMessage.QOSType.MOST_ONE); + store.add(financeOne); + assertTrue(store.matches("finance/stock").contains(financeOne)); + } + + @Test + public void testMatchManySingle() { + Subscription manySub = new Subscription(null, "+/+", AbstractMessage.QOSType.MOST_ONE); + store.add(manySub); + + //verify + assertTrue(store.matches("/finance").contains(manySub)); + } + + + @Test + public void testMatchSlashSingle() { + Subscription slashPlusSub = new Subscription(null, "/+", AbstractMessage.QOSType.MOST_ONE); + store.add(slashPlusSub); + Subscription anySub = new Subscription(null, "+", AbstractMessage.QOSType.MOST_ONE); + store.add(anySub); + + //Verify + assertEquals(1, store.matches("/finance").size()); + assertTrue(store.matches("/finance").contains(slashPlusSub)); + assertFalse(store.matches("/finance").contains(anySub)); + } + + + @Test + public void testMatchManyDeepSingle() { + Subscription slashPlusSub = new Subscription(null, "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE); + store.add(slashPlusSub); + + Subscription slashPlusDeepSub = new Subscription(null, "/+/stock/+", AbstractMessage.QOSType.MOST_ONE); + store.add(slashPlusDeepSub); + + //Verify + assertTrue(store.matches("/finance/stock/ibm").containsAll(Arrays.asList(slashPlusSub, slashPlusDeepSub))); + } + + @Test + public void testMatchSimpleMulti_allTheTree() { + store.add(new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE)); + assertFalse(store.matches("finance").isEmpty()); + assertFalse(store.matches("finance/ibm").isEmpty()); + } + + @Test + public void testMatchSimpleMulti_zeroLevel() { + //check MULTI in case of zero level match + store.add(new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE)); + assertFalse(store.matches("finance").isEmpty()); + } + + private static Token[] asArray(Object... l) { Token[] tokens = new Token[l.length]; for (int i = 0; i < l.length; i++) { Object o = l[i]; if (o instanceof Token) { - tokens[i] = (Token)o; + tokens[i] = (Token) o; } else { tokens[i] = new Token(o.toString()); } } - + return tokens; } - + private void assertEqualsSeq(Token[] exptected, List result) { List expectedList = Arrays.asList(exptected); assertEquals(expectedList, result);