anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added raw implementation for subscriptions persistence

Browse Source
This commit is contained in:
selva.andre
2012-08-10 23:24:59 +02:00
parent 943a962bb9
commit 389c906d8f
4 changed files with 171 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/pom.xml
+15 -2
View File
@@ -1,6 +1,6 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<relativePath>../pom.xml</relativePath>
@@ -25,7 +25,7 @@
<version>2.0.4</version>
</dependency>
<!-- <dependency>
<!-- <dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-jdk14</artifactId>
<version>1.6.4</version>
@@ -37,6 +37,12 @@
<artifactId>slf4j-log4j12</artifactId>
<version>1.6.4</version>
</dependency>
<dependency>
<groupId>org.fusesource.hawtdb</groupId>
<artifactId>hawtdb</artifactId>
<version>1.6</version>
</dependency>
<dependency>
<groupId>junit</groupId>
@@ -77,4 +83,11 @@
</plugin>
</plugins>
</build>
<!-- <repositories>
<repository>
<id>Fusesource Releases</id>
<url>http://repo.fusesource.com/nexus/content/repositories/snapshots</url>
</repository>
</repositories>-->
</project>
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+5
View File
@@ -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;
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java
+116 -50
View File
@@ -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<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;
}
@@ -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<Subscription> subscriptions() {
return m_subscriptions;
}
void matches(Queue<Token> tokens, List<Subscription> 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<Subscription> subsToRemove = new ArrayList<Subscription>();
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<Subscription> subscriptions = new ArrayList<Subscription>();
private TreeNode subscriptions = new TreeNode(null);
private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
//pesistent Map of clientID, list of Subscriptions
private SortedIndex<String, List<Subscription>> 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<String, List<Subscription>> indexFactory =
// new HashIndexFactory<String, List<Subscription>>();
BTreeIndexFactory<String, List<Subscription>> indexFactory =
new BTreeIndexFactory<String, List<Subscription>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_persistent = indexFactory.openOrCreate(pageFile);
//reaload any subscriptions persisted
LOG.debug("Reloading all stored subscriptions...");
for(Map.Entry<String, List<Subscription>> entry: m_persistent) {
for (Subscription subscription : entry.getValue()) {
addDirect(subscription);
}
}
LOG.debug("Finished loading");
}
protected void addDirect(Subscription newSubscription) {
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);
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<Subscription>());
}
List<Subscription> 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<Subscription> 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<Token> tokenQueue = new LinkedBlockingDeque<Token>(tokens);
List<Subscription> matchingSubs = new ArrayList<Subscription>();
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<Token> splitTopic(String topic) throws ParseException {
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++) {
for (int i = 0; i < splitted.length; i++) {
String s = splitted[i];
if (s.isEmpty()) {
if (i != 0){
if (i != 0) {
throw new ParseException("Bad format of topic, expetec topic name between separators", i);
}
res.add(Token.EMPTY);
} else if (s.equals("#")) {
//check that multi is the last symbol
if (i != splitted.length - 1 ) {
if (i != splitted.length - 1) {
throw new ParseException("Bad format of topic, the multi symbol (#) has to be the last one after a separator", i);
}
res.add(Token.MULTI);
@@ -279,7 +345,7 @@ public class SubscriptionsStore {
res.add(new Token(s));
}
}
return res;
}
}
parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java
+35 -35
View File
@@ -155,40 +155,40 @@ public class ServerIntegrationTest {
}
// @Test
// public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws IOException {
// Client client = new Client("localhost", Server.PORT, "CLID_123");
// client.connect(false); //without session cleanup
//
// client.subscribe("/topic", new IPublishCallback() {
//
// public void published(String topic, byte[] message) {
// received = true;
// }
// });
// client.close();
// client.shutdown();
// server.stopServer();
//
// server.startServer();
//
// //Exercise that the client maintain the subscriptions
// client = new Client("localhost", Server.PORT, "CLID_123");
// client.register("/topic", new IPublishCallback() {
//
// public void published(String topic, byte[] message) {
// received = true;
// }
// });
// client.connect(false);
// client.publish("/topic", "Test my payload".getBytes());
// client.close();
//
// //Verify
// assertTrue(received);
//
// //TearDown
// client.shutdown();
// }
@Test
public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws IOException {
Client client = new Client("localhost", Server.PORT, "CLID_123");
client.connect(false); //without session cleanup
client.subscribe("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
received = true;
}
});
client.close();
client.shutdown();
server.stopServer();
server.startServer();
//Exercise that the client maintain the subscriptions
client = new Client("localhost", Server.PORT, "CLID_123");
client.register("/topic", new IPublishCallback() {
public void published(String topic, byte[] message) {
received = true;
}
});
client.connect(false);
client.publish("/topic", "Test my payload".getBytes());
client.close();
//Verify
assertTrue(received);
//TearDown
client.shutdown();
}
}