From fdedfecfba2b2a999b5813b2a0a32c425d2436f7 Mon Sep 17 00:00:00 2001
From: "selva.andre" <andrea@sempai>
Date: Tue, 7 Aug 2012 00:17:47 +0200
Subject: [PATCH] Added first implementation for topic tree subscriptions.

---
 .../spi/impl/SubscriptionsStore.java          | 143 ++++++++++++++++--
 .../spi/impl/SubscriptionsStoreTest.java      |  22 +++
 2 files changed, 154 insertions(+), 11 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 fcf68e4c..b81c29e5 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
@@ -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()) {
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 d06b2428..5bcbf0bd 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
@@ -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];