Added first implementation for topic tree subscriptions.

2012-08-07 00:17:47 +02:00
parent f8d75bf004
commit fdedfecfba
2 changed files with 154 additions and 11 deletions
@@ -2,7 +2,10 @@ package org.dna.mqtt.moquette.messaging.spi.impl;

 import java.text.ParseException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;

 /**
 * Represents a tree of published topics
@@ -11,6 +14,88 @@ import java.util.List;
 */
 public class SubscriptionsStore {
    
+    private class TreeNode {
+        TreeNode m_parent;
+        Token m_token;
+        List<TreeNode> m_children = new ArrayList<TreeNode>();
+        List<Subscription> m_subscriptions = new ArrayList<Subscription>();
+        
+        TreeNode(TreeNode parent) {
+            this.m_parent = parent;
+        }
+        
+        Token getToken() {
+            return m_token;
+        }
+
+        void setToken(Token topic) {
+            this.m_token = topic;
+        }
+        
+        void addSubcription(Subscription s) {
+            //avoid double registering
+            if (m_subscriptions.contains(s)) {
+                return;
+            }
+            m_subscriptions.add(s);
+        }
+        
+        void addChild(TreeNode child) {
+            m_children.add(child);
+        }
+        
+        boolean isLeaf() {
+            return m_children.isEmpty();
+        }
+        
+        
+        /**
+         * Search for children that has the specified token, if not found return
+         * null;
+         */
+        TreeNode childWithToken(Token token) {
+            for (TreeNode child : m_children) {
+                if (child.getToken().equals(token)) {
+                    return child;
+                }
+            }
+            
+            return null;
+        }
+        
+        List<Subscription> subscriptions() {
+            return m_subscriptions;
+        }
+        
+        
+        /**
+         * Return next matching node on the tree path, or null if not found any 
+         * match.
+         */
+        TreeNode matchNext(Token token) {
+            //quite similar to childWithToken
+            //TODO this should take care of # and + management
+            for (TreeNode child : m_children) {
+                if (child.getToken().equals(token)) {
+                    return child;
+                }
+            }
+            
+            return null;
+        }
+        
+        /**
+         * Return the number of registered subscriptions
+         */
+        int size() {
+            int res = m_subscriptions.size();
+            for (TreeNode child : m_children) {
+                res += child.size();
+            }
+            return res;
+        }
+    }
+    
    protected static class Token {
    
        static final Token EMPTY = new Token("");
@@ -50,29 +135,61 @@ public class SubscriptionsStore {
        }
    }

-    private List<Subscription> subscriptions = new ArrayList<Subscription>();
+//    private List<Subscription> subscriptions = new ArrayList<Subscription>();
+    private TreeNode subscriptions = new TreeNode(null);

    public void add(Subscription newSubscription) {
-        if (subscriptions.contains(newSubscription)) {
-            return;
+        List<Token> tokens = new ArrayList<Token>();
+        try {
+            tokens = splitTopic(newSubscription.topic);
+        } catch (ParseException ex) {
+            //TODO handle the parse exception
+            Logger.getLogger(SubscriptionsStore.class.getName()).log(Level.SEVERE, null, ex);
        }
-        subscriptions.add(newSubscription);
+        
+        TreeNode current = subscriptions;
+        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 {
+                //create a new node for the newly inserted token
+                matchingChildren = new TreeNode(current);
+                matchingChildren.setToken(token);
+                current = matchingChildren;
+            }
+        }
+        current.addSubcription(newSubscription);
    }

    public List<Subscription> matches(String topic) {
-        List<Subscription> m = new ArrayList<Subscription>();
-        for (Subscription sub : subscriptions) {
-            if (sub.match(topic)) {
-                m.add(sub);
+        List<Token> tokens = new ArrayList<Token>();
+        try {
+            tokens = splitTopic(topic);
+        } catch (ParseException ex) {
+            //TODO handle the parse exception
+            Logger.getLogger(SubscriptionsStore.class.getName()).log(Level.SEVERE, null, ex);
+        }
+        TreeNode current = subscriptions;
+        for (Token token : tokens) {
+            current = current.matchNext(token);
+            if (current == null) {
+                break;
            }
        }
-        return m;
+        if (current != null) {
+            return current.subscriptions();
+        } else {
+            return Collections.EMPTY_LIST;
+        }
    }

    public boolean contains(Subscription sub) {
-        return subscriptions.contains(sub);
+        return !matches(sub.topic).isEmpty();
    }
-
+    
    public int size() {
        return subscriptions.size();
    }
@@ -81,6 +198,10 @@ public class SubscriptionsStore {
        List res = new ArrayList<Token>();
        String[] splitted = topic.split("/");
        
+        if (splitted.length == 0) {
+            res.add(Token.EMPTY);
+        }
+        
        for (int i=0; i<splitted.length; i++) {
            String s = splitted[i];
            if (s.isEmpty()) {
@@ -4,6 +4,7 @@ import java.text.ParseException;
 import java.util.Arrays;
 import java.util.List;
 import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token;
+import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import static org.junit.Assert.*;
 import org.junit.Before;
 import org.junit.Test;
@@ -31,6 +32,9 @@ public class SubscriptionsStoreTest {
        
        tokens = store.splitTopic("/finance/stock/ibm");
        assertEqualsSeq(asArray(Token.EMPTY, "finance", "stock", "ibm"), tokens);
+        
+        tokens = store.splitTopic("/");
+        assertEqualsSeq(asArray(Token.EMPTY), tokens);
    }
    
    
@@ -77,6 +81,24 @@ public class SubscriptionsStoreTest {
        store.splitTopic("finance+");
    }
    
+    @Test
+    public void testMatchSimple() {
+        store.add(new Subscription(null, "/", AbstractMessage.QOSType.MOST_ONE));
+        assertTrue(store.matches("finance").isEmpty());
+        
+        store.add(new Subscription(null, "/finance", AbstractMessage.QOSType.MOST_ONE));
+        assertTrue(store.matches("finance").isEmpty());
+    }
+    
+    @Test
+    public void testMatchSimpleMulti() {
+        store.add(new Subscription(null, "#", AbstractMessage.QOSType.MOST_ONE));
+        assertFalse(store.matches("finance").isEmpty());
+        
+        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];