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

Moved creation of messsaging index store from SubscriptionsStore up to SimpleMessaging, to be used also for retained messages

Browse Source
This commit is contained in:
selva.andre
2012-08-14 22:52:53 +02:00
parent 025b5c77ff
commit d2a667ef9e
2 changed files with 129 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+108 -6
View File
@@ -1,26 +1,92 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.io.File;
import java.io.IOException;
import java.text.ParseException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.logging.Level;
import org.dna.mqtt.moquette.MQTTException;
import org.dna.mqtt.moquette.messaging.spi.IMessaging;
import org.dna.mqtt.moquette.messaging.spi.INotifier;
import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token;
import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType;
import org.dna.mqtt.moquette.server.Server;
import org.fusesource.hawtbuf.codec.StringCodec;
import org.fusesource.hawtdb.api.BTreeIndexFactory;
import org.fusesource.hawtdb.api.MultiIndexFactory;
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;
/**
*
* @author andrea
*/
public class SimpleMessaging implements IMessaging {
private static class StoredMessage {
QOSType m_qos;
byte[] m_payload;
private StoredMessage(byte[] message, QOSType qos) {
m_qos = qos;
m_payload = message;
}
QOSType getQos() {
return m_qos;
}
byte[] getPayload() {
return m_payload;
}
}
private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class);
//TODO is poorly performace on concurrent load, use at least ReadWriteLock
private ReadWriteLock rwLock = new ReentrantReadWriteLock();
private SubscriptionsStore subscriptions = new SubscriptionsStore();
private INotifier m_notifier;
private MultiIndexFactory m_multiIndexFactory;
private PageFileFactory pageFactory;
private SortedIndex<String, StoredMessage> m_retainedStore;
public SimpleMessaging() {
subscriptions.init();
String storeFile = Server.STORAGE_FILE_PATH;
pageFactory = new PageFileFactory();
File tmpFile;
try {
tmpFile = new File(storeFile);
tmpFile.createNewFile();
} catch (IOException ex) {
LOG.error(null, ex);
throw new MQTTException("Can't create temp file for subscriptions storage [" + storeFile + "]", ex);
}
pageFactory.setFile(tmpFile);
pageFactory.open();
PageFile pageFile = pageFactory.getPageFile();
m_multiIndexFactory = new MultiIndexFactory(pageFile);
subscriptions.init(m_multiIndexFactory);
initRetainedMessageStore();
}
private void initRetainedMessageStore() {
BTreeIndexFactory<String, StoredMessage> indexFactory = new BTreeIndexFactory<String, StoredMessage>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
// m_persistent = indexFactory.openOrCreate(pageFile);
m_retainedStore = (SortedIndex<String, StoredMessage>) m_multiIndexFactory.openOrCreate("retained", indexFactory);
}
public void setNotifier(INotifier notifier) {
@@ -28,14 +94,13 @@ public class SimpleMessaging implements IMessaging {
}
public void publish(String topic, byte[] message, QOSType qos, boolean retain) {
//TODO for this moment no retain management and no qos > 0
//trivial implementation
//TODO for this moment no qos > 0 trivial implementation
QOSType defQos = QOSType.MOST_ONE;
rwLock.readLock().lock();
try {
for (Subscription sub : subscriptions.matches(topic)) {
m_notifier.notify(sub.clientId, topic, defQos, message);
m_notifier.notify(sub.clientId, topic, defQos, message, false);
}
} finally {
rwLock.readLock().unlock();
@@ -49,6 +114,38 @@ public class SimpleMessaging implements IMessaging {
rwLock.writeLock().unlock();
}
/**
* Verify if the 2 topics matching respecting the rules of MQTT Appendix A
*/
protected boolean matchTopics(String msgTopic, String subscriptionTopic) {
try {
List<Token> msgTokens = SubscriptionsStore.splitTopic(msgTopic);
List<Token> subscriptionTokens = SubscriptionsStore.splitTopic(subscriptionTopic);
int i = 0;
for (; i< subscriptionTokens.size(); i++) {
Token subToken = subscriptionTokens.get(i);
if (subToken != Token.MULTI && subToken != Token.SINGLE) {
Token msgToken = msgTokens.get(i);
if (!msgToken.equals(subToken)) {
return false;
}
} else {
if (subToken == Token.MULTI) {
return true;
}
if (subToken == Token.SINGLE) {
//skip a step forward
}
}
}
return i == msgTokens.size();
} catch (ParseException ex) {
LOG.error(null, ex);
throw new RuntimeException(ex);
}
}
protected SubscriptionsStore getSubscriptions() {
return subscriptions;
}
@@ -61,7 +158,12 @@ public class SimpleMessaging implements IMessaging {
public void close() {
rwLock.writeLock().lock();
subscriptions.close();
rwLock.writeLock().unlock();
try {
pageFactory.close();
} catch (IOException ex) {
LOG.error(null, ex);
} finally {
rwLock.writeLock().unlock();
}
}
}
parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java
+21 -33
View File
@@ -1,6 +1,5 @@
package org.dna.mqtt.moquette.messaging.spi.impl;
import java.io.File;
import java.io.IOException;
import java.text.ParseException;
import java.util.ArrayList;
@@ -9,14 +8,9 @@ import java.util.Map;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.logging.Level;
import org.dna.mqtt.moquette.MQTTException;
import org.dna.mqtt.moquette.server.MQTTHandler;
import org.dna.mqtt.moquette.server.Server;
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.MultiIndexFactory;
import org.fusesource.hawtdb.api.SortedIndex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -203,30 +197,30 @@ public class SubscriptionsStore {
//pesistent Map of clientID, list of Subscriptions
private SortedIndex<String, List<Subscription>> m_persistent;
// private PageFile pageFile;
private PageFileFactory factory;
// private PageFileFactory factory;
/**
* Initialize basic store structures, like the FS storage to maintain
* client's topics subscriptions
*/
public void init() {
public void init(MultiIndexFactory multiFactory) {
/*StringBuilder storeFile = new StringBuilder();
storeFile.append(System.getProperty("user.home"))
.append(File.separator).append("moquette_store.hawtdb");*/
String storeFile = Server.STORAGE_FILE_PATH;
// String storeFile = Server.STORAGE_FILE_PATH;
factory = new PageFileFactory();
File tmpFile;
try {
tmpFile = new File(storeFile.toString());
tmpFile.createNewFile();
} catch (IOException ex) {
LOG.error(null, ex);
throw new MQTTException("Can't create temp file for subscriptions storage [" + storeFile + "]", ex);
}
factory.setFile(/*new File("mydb.dat")*/tmpFile);
factory.open();
PageFile pageFile = factory.getPageFile();
// factory = new PageFileFactory();
// File tmpFile;
// try {
// tmpFile = new File(storeFile.toString());
// tmpFile.createNewFile();
// } catch (IOException ex) {
// LOG.error(null, ex);
// throw new MQTTException("Can't create temp file for subscriptions storage [" + storeFile + "]", ex);
// }
// factory.setFile(/*new File("mydb.dat")*/tmpFile);
// factory.open();
// PageFile pageFile = factory.getPageFile();
// HashIndexFactory<String, List<Subscription>> indexFactory =
// new HashIndexFactory<String, List<Subscription>>();
@@ -234,7 +228,8 @@ public class SubscriptionsStore {
new BTreeIndexFactory<String, List<Subscription>>();
indexFactory.setKeyCodec(StringCodec.INSTANCE);
m_persistent = indexFactory.openOrCreate(pageFile);
// m_persistent = indexFactory.openOrCreate(pageFile);
m_persistent = (SortedIndex<String, List<Subscription>>) multiFactory.openOrCreate("subscriptions", indexFactory);
//reaload any subscriptions persisted
LOG.debug("Reloading all stored subscriptions...");
@@ -246,15 +241,6 @@ public class SubscriptionsStore {
LOG.debug("Finished loading");
}
public void close() {
try {
factory.close();
} catch (IOException ex) {
//TODO handle probably firing
LOG.error(null, ex);
}
}
protected void addDirect(Subscription newSubscription) {
List<Token> tokens = new ArrayList<Token>();
try {
@@ -330,7 +316,9 @@ public class SubscriptionsStore {
return subscriptions.size();
}
protected List<Token> splitTopic(String topic) throws ParseException {
//TODO this should be static
protected static List<Token> splitTopic(String topic) throws ParseException {
List res = new ArrayList<Token>();
String[] splitted = topic.split("/");