diff --git a/parser/pom.xml b/parser/pom.xml
index 3ee0f88e..8c1904b4 100644
--- a/parser/pom.xml
+++ b/parser/pom.xml
@@ -1,6 +1,6 @@
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
4.0.0
../pom.xml
@@ -25,7 +25,7 @@
2.0.4
-
diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
index 59be5b38..48a1f323 100644
--- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
@@ -18,6 +18,11 @@ public class SimpleMessaging implements IMessaging {
private INotifier m_notifier;
+
+ public SimpleMessaging() {
+ subscriptions.init();
+ }
+
public void setNotifier(INotifier notifier) {
m_notifier = notifier;
}
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 682c9d9e..d87e39df 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
@@ -1,13 +1,23 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
+import java.io.File;
+import java.io.IOException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Map;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
-import java.util.logging.Level;
-import java.util.logging.Logger;
+import org.dna.mqtt.moquette.MQTTException;
+import org.dna.mqtt.moquette.server.MQTTHandler;
+import org.fusesource.hawtbuf.codec.StringCodec;
+import org.fusesource.hawtdb.api.BTreeIndexFactory;
+import org.fusesource.hawtdb.api.PageFile;
+import org.fusesource.hawtdb.api.PageFileFactory;
+import org.fusesource.hawtdb.api.SortedIndex;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Represents a tree of published topics
@@ -17,15 +27,16 @@ import java.util.logging.Logger;
public class SubscriptionsStore {
private class TreeNode {
+
TreeNode m_parent;
Token m_token;
List m_children = new ArrayList();
List m_subscriptions = new ArrayList();
-
+
TreeNode(TreeNode parent) {
this.m_parent = parent;
}
-
+
Token getToken() {
return m_token;
}
@@ -33,7 +44,7 @@ public class SubscriptionsStore {
void setToken(Token topic) {
this.m_token = topic;
}
-
+
void addSubcription(Subscription s) {
//avoid double registering
if (m_subscriptions.contains(s)) {
@@ -41,16 +52,15 @@ public class SubscriptionsStore {
}
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;
@@ -61,18 +71,17 @@ public class SubscriptionsStore {
return child;
}
}
-
+
return null;
}
-
+
List subscriptions() {
return m_subscriptions;
}
-
-
+
void matches(Queue tokens, List matchingSubs) {
Token t = tokens.poll();
-
+
//check if t is null <=> tokens finished
if (t == null) {
matchingSubs.addAll(m_subscriptions);
@@ -82,16 +91,16 @@ public class SubscriptionsStore {
matchingSubs.addAll(n.subscriptions());
}
}
-
+
return;
}
-
+
//we are on MULTI, than add subscriptions and return
if (m_token == Token.MULTI) {
- matchingSubs.addAll(m_subscriptions);
- return;
+ matchingSubs.addAll(m_subscriptions);
+ return;
}
-
+
for (TreeNode n : m_children) {
if (n.getToken().match(t)) {
//Create a copy of token, alse if navigate 2 sibling it
@@ -100,7 +109,7 @@ public class SubscriptionsStore {
}
}
}
-
+
/**
* Return the number of registered subscriptions
*/
@@ -115,51 +124,50 @@ public class SubscriptionsStore {
void removeClientSubscriptions(String clientID) {
//collect what to delete and then delete to avoid ConcurrentModification
List subsToRemove = new ArrayList();
- for(Subscription s : m_subscriptions) {
+ for (Subscription s : m_subscriptions) {
if (s.clientId.equals(clientID)) {
subsToRemove.add(s);
}
}
-
+
for (Subscription s : subsToRemove) {
- m_subscriptions.remove(s);
+ m_subscriptions.remove(s);
}
-
+
//go deep
for (TreeNode child : m_children) {
child.removeClientSubscriptions(clientID);
}
}
}
-
+
protected static class Token {
-
+
static final Token EMPTY = new Token("");
static final Token MULTI = new Token("#");
static final Token SINGLE = new Token("+");
-
String name;
-
+
protected Token(String s) {
name = s;
}
-
+
protected String name() {
return name;
}
-
+
protected boolean match(Token t) {
if (t == MULTI || t == SINGLE) {
return false;
}
-
+
if (this == MULTI || this == SINGLE) {
return true;
}
-
+
return equals(t);
}
-
+
@Override
public int hashCode() {
int hash = 7;
@@ -181,29 +189,70 @@ public class SubscriptionsStore {
}
return true;
}
-
+
@Override
public String toString() {
return name;
}
}
-
// private List subscriptions = new ArrayList();
private TreeNode subscriptions = new TreeNode(null);
+
+ private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
+
+ //pesistent Map of clientID, list of Subscriptions
+ private SortedIndex> m_persistent;
+ private PageFile pageFile;
- public void add(Subscription newSubscription) {
+ /**
+ * Initialize basic store structures, like the FS storage to maintain
+ * client's topics subscriptions
+ */
+ public void init() {
+ PageFileFactory factory = new PageFileFactory();
+ File tmpFile;
+ try {
+ tmpFile = File.createTempFile("hawtdb", "test");
+ } catch (IOException ex) {
+ LOG.error(null, ex);
+ throw new MQTTException("Can't create temp file for subscriptions storage", ex);
+ }
+ factory.setFile(/*new File("mydb.dat")*/tmpFile);
+ factory.open();
+ pageFile = factory.getPageFile();
+
+// HashIndexFactory> indexFactory =
+// new HashIndexFactory>();
+ BTreeIndexFactory> indexFactory =
+ new BTreeIndexFactory>();
+ indexFactory.setKeyCodec(StringCodec.INSTANCE);
+
+ m_persistent = indexFactory.openOrCreate(pageFile);
+
+ //reaload any subscriptions persisted
+ LOG.debug("Reloading all stored subscriptions...");
+ for(Map.Entry> entry: m_persistent) {
+ for (Subscription subscription : entry.getValue()) {
+ addDirect(subscription);
+ }
+ }
+ LOG.debug("Finished loading");
+ }
+
+
+ protected void addDirect(Subscription newSubscription) {
List tokens = new ArrayList();
try {
tokens = splitTopic(newSubscription.topic);
} catch (ParseException ex) {
//TODO handle the parse exception
- Logger.getLogger(SubscriptionsStore.class.getName()).log(Level.SEVERE, null, ex);
+ LOG.error(null, ex);
}
-
+
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;
@@ -217,12 +266,29 @@ public class SubscriptionsStore {
}
current.addSubcription(newSubscription);
}
-
+
+ public void add(Subscription newSubscription) {
+ addDirect(newSubscription);
+
+ //log the subscription
+ String clientID = newSubscription.getClientId();
+ if (!m_persistent.containsKey(clientID)) {
+ m_persistent.put(clientID, new ArrayList());
+ }
+
+ List subs = m_persistent.get(clientID);
+ subs.add(newSubscription);
+ pageFile.flush();
+ }
+
/**
* Visit the topics tree to remove matching subscriptions with clientID
*/
public void removeForClient(String clientID) {
subscriptions.removeClientSubscriptions(clientID);
+
+ //remove from log all subscriptions
+ m_persistent.remove(clientID);
}
public List matches(String topic) {
@@ -231,9 +297,9 @@ public class SubscriptionsStore {
tokens = splitTopic(topic);
} catch (ParseException ex) {
//TODO handle the parse exception
- Logger.getLogger(SubscriptionsStore.class.getName()).log(Level.SEVERE, null, ex);
+ LOG.error(null, ex);
}
-
+
Queue tokenQueue = new LinkedBlockingDeque(tokens);
List matchingSubs = new ArrayList();
subscriptions.matches(tokenQueue, matchingSubs);
@@ -243,29 +309,29 @@ public class SubscriptionsStore {
public boolean contains(Subscription sub) {
return !matches(sub.topic).isEmpty();
}
-
+
public int size() {
return subscriptions.size();
}
-
+
protected List splitTopic(String topic) throws ParseException {
List res = new ArrayList();
String[] splitted = topic.split("/");
-
+
if (splitted.length == 0) {
res.add(Token.EMPTY);
}
-
- for (int i=0; i