From d2a667ef9ebd25543b31bc7456b50ea518e763ed Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Tue, 14 Aug 2012 22:52:53 +0200 Subject: [PATCH] Moved creation of messsaging index store from SubscriptionsStore up to SimpleMessaging, to be used also for retained messages --- .../messaging/spi/impl/SimpleMessaging.java | 114 +++++++++++++++++- .../spi/impl/SubscriptionsStore.java | 54 ++++----- 2 files changed, 129 insertions(+), 39 deletions(-) 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 d46fd5be..7175f23f 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 @@ -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 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 indexFactory = new BTreeIndexFactory(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + +// m_persistent = indexFactory.openOrCreate(pageFile); + m_retainedStore = (SortedIndex) 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 msgTokens = SubscriptionsStore.splitTopic(msgTopic); + List 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(); + } } } 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 a6e3b729..481282c0 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,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> 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> indexFactory = // new HashIndexFactory>(); @@ -234,7 +228,8 @@ public class SubscriptionsStore { new BTreeIndexFactory>(); indexFactory.setKeyCodec(StringCodec.INSTANCE); - m_persistent = indexFactory.openOrCreate(pageFile); +// m_persistent = indexFactory.openOrCreate(pageFile); + m_persistent = (SortedIndex>) 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 tokens = new ArrayList(); try { @@ -330,7 +316,9 @@ public class SubscriptionsStore { return subscriptions.size(); } - protected List splitTopic(String topic) throws ParseException { + + //TODO this should be static + protected static List splitTopic(String topic) throws ParseException { List res = new ArrayList(); String[] splitted = topic.split("/");