diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMatchingCondition.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMatchingCondition.java new file mode 100644 index 00000000..21a16ccb --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMatchingCondition.java @@ -0,0 +1,8 @@ +package org.dna.mqtt.moquette.messaging.spi; + +/** + */ +public interface IMatchingCondition { + boolean match(String key); +} + diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java new file mode 100644 index 00000000..0ee6b843 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java @@ -0,0 +1,43 @@ +package org.dna.mqtt.moquette.messaging.spi; + +import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; +import org.dna.mqtt.moquette.messaging.spi.impl.Subscription; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging.StoredMessage; + +import java.util.Collection; +import java.util.List; + +/** + * Defines the SPI to be implemented by a StorageService that handle persistence of messages and subscriptions. + */ +public interface IStorageService { + + /** + * Used to initialize all persistent store structures + * */ + void initStore(); + + void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos); + + Collection searchMatching(IMatchingCondition condition); + + void storePublishForFuture(PublishEvent evt); + + List retrivePersistedPublishes(String clientID); + + void cleanPersistedPublishes(String clientID); + + void cleanInFlight(String msgID); + + void addInFlight(PublishEvent evt, String publishKey); + + void addNewSubscription(Subscription newSubscription, String clientID); + + void removeAllSubscriptions(String clientID); + + List retrieveAllSubscriptions(); + + void close(); +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java new file mode 100644 index 00000000..85848203 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -0,0 +1,187 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import org.dna.mqtt.moquette.MQTTException; +import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; +import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import org.dna.mqtt.moquette.server.Server; +import org.fusesource.hawtbuf.codec.StringCodec; +import org.fusesource.hawtdb.api.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; + +import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging.StoredMessage; + + +/** + * Implementation of IStorageService backed by HawtDB + */ +public class HawtDBStorageService implements IStorageService { + + private static final Logger LOG = LoggerFactory.getLogger(HawtDBStorageService.class); + + private MultiIndexFactory m_multiIndexFactory; + private PageFileFactory pageFactory; + + //maps clientID to the list of pending messages stored + private SortedIndex> m_persistentMessageStore; + private SortedIndex m_retainedStore; + //bind clientID+MsgID -> evt message published + private SortedIndex m_inflightStore; + + //persistent Map of clientID, list of Subscriptions + private SortedIndex> m_persistentSubscriptions; + + public HawtDBStorageService() { + 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); + } + + + public void initStore() { + initRetainedStore(); + //init the message store for QoS 1/2 messages in clean sessions + initPersistentMessageStore(); + initInflightMessageStore(); + initPersistentSubscriptions(); + } + + private void initRetainedStore() { + BTreeIndexFactory indexFactory = new BTreeIndexFactory(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_retainedStore = (SortedIndex) m_multiIndexFactory.openOrCreate("retained", indexFactory); + + } + + + private void initPersistentMessageStore() { + BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); + } + + private void initPersistentSubscriptions() { + BTreeIndexFactory> indexFactory = + new BTreeIndexFactory>(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_persistentSubscriptions = (SortedIndex>) m_multiIndexFactory.openOrCreate("subscriptions", indexFactory); + } + + /** + * Initialize the message store used to handle the temporary storage of QoS 1,2 + * messages in flight. + */ + private void initInflightMessageStore() { + BTreeIndexFactory indexFactory = new BTreeIndexFactory(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_inflightStore = (SortedIndex) m_multiIndexFactory.openOrCreate("inflight", indexFactory); + } + + public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { + if (message.length == 0) { + //clean the message from topic + m_retainedStore.remove(topic); + } else { + //store the message to the topic + m_retainedStore.put(topic, new StoredMessage(message, qos)); + } + } + + public Collection searchMatching(IMatchingCondition condition) { + LOG.debug("searchMatching scanning all retained messages, presents are " + m_retainedStore.size()); + + List results = new ArrayList(); + + for (Map.Entry entry : m_retainedStore) { + StoredMessage storedMsg = entry.getValue(); + if (condition.match(entry.getKey())) { + results.add(storedMsg); + } + } + + return results; + } + + public void storePublishForFuture(PublishEvent evt) { + List storedEvents; + String clientID = evt.getClientID(); + if (!m_persistentMessageStore.containsKey(clientID)) { + storedEvents = new ArrayList(); + } else { + storedEvents = m_persistentMessageStore.get(clientID); + } + storedEvents.add(evt); + m_persistentMessageStore.put(clientID, storedEvents); + } + + public List retrivePersistedPublishes(String clientID) { + return m_persistentMessageStore.get(clientID); + } + + public void cleanPersistedPublishes(String clientID) { + m_persistentMessageStore.remove(clientID); + } + + public void cleanInFlight(String msgID) { + m_inflightStore.remove(msgID); + } + + public void addInFlight(PublishEvent evt, String publishKey) { + m_inflightStore.put(publishKey, evt); + } + + public void addNewSubscription(Subscription newSubscription, String clientID) { + if (!m_persistentSubscriptions.containsKey(clientID)) { + m_persistentSubscriptions.put(clientID, new ArrayList()); + } + + List subs = m_persistentSubscriptions.get(clientID); + subs.add(newSubscription); + m_persistentSubscriptions.put(clientID, subs); + } + + public void removeAllSubscriptions(String clientID) { + m_persistentSubscriptions.remove(clientID); + } + + public List retrieveAllSubscriptions() { + List allSubscriptions = new ArrayList(); + for (Map.Entry> entry : m_persistentSubscriptions) { + allSubscriptions.addAll(entry.getValue()); + } + return allSubscriptions; + } + + public void close() { + try { + pageFactory.close(); + } catch (IOException ex) { + LOG.error(null, ex); + } + } +} 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 22ffc24e..e8b13637 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,43 +1,43 @@ package org.dna.mqtt.moquette.messaging.spi.impl; -import java.io.File; -import java.io.IOException; -import java.io.Serializable; -import java.text.ParseException; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.LinkedBlockingQueue; import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.MQTTException; +import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.INotifier; +import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.SubscriptionsStore.Token; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.server.Constants; 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; +import java.io.File; +import java.io.IOException; +import java.io.Serializable; +import java.text.ParseException; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; + /** * * @author andrea */ public class SimpleMessaging implements IMessaging, Runnable { - private static class StoredMessage implements Serializable { + //TODO probably move this + public static class StoredMessage implements Serializable { QOSType m_qos; byte[] m_payload; - private StoredMessage(byte[] message, QOSType qos) { + StoredMessage(byte[] message, QOSType qos) { m_qos = qos; m_payload = message; } @@ -58,64 +58,17 @@ public class SimpleMessaging implements IMessaging, Runnable { private BlockingQueue m_inboundQueue = new LinkedBlockingQueue(); private INotifier m_notifier; - private MultiIndexFactory m_multiIndexFactory; - private PageFileFactory pageFactory; - private SortedIndex m_retainedStore; - //bind clientID+MsgID -> evt message published - private SortedIndex m_inflightStore; - //maps clientID to the list of pending messages stored - private SortedIndex> m_persistentMessageStore; + + private IStorageService m_storageService; public SimpleMessaging() { - 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(); - initInflightMessageStore(); - //init the message store for QoS 1/2 messages in clean sessions - initPersistentMessageStore(); + m_storageService = new HawtDBStorageService(); + m_storageService.initStore(); + + subscriptions.init(m_storageService); } + - private void initRetainedMessageStore() { - BTreeIndexFactory indexFactory = new BTreeIndexFactory(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); - -// m_persistent = indexFactory.openOrCreate(pageFile); - m_retainedStore = (SortedIndex) m_multiIndexFactory.openOrCreate("retained", indexFactory); - } - - /** - * Initialize the message store used to handle the temporary storage of QoS 1,2 - * messages in flight. - */ - private void initInflightMessageStore() { - BTreeIndexFactory indexFactory = new BTreeIndexFactory(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); - - m_inflightStore = (SortedIndex) m_multiIndexFactory.openOrCreate("inflight", indexFactory); - } - - private void initPersistentMessageStore() { - BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); - - m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); - } - public void setNotifier(INotifier notifier) { m_notifier= notifier; } @@ -258,7 +211,7 @@ public class SimpleMessaging implements IMessaging, Runnable { processDisconnect((DisconnectEvent) evt); } else if (evt instanceof CleanInFlightEvent) { //remove the message from inflight storage - m_inflightStore.remove(((CleanInFlightEvent)evt).getMsgId()); + m_storageService.cleanInFlight(((CleanInFlightEvent) evt).getMsgId()); } else if (evt instanceof RepublishEvent) { processRepublish((RepublishEvent) evt); } @@ -281,7 +234,7 @@ public class SimpleMessaging implements IMessaging, Runnable { if (qos == QOSType.LEAST_ONE) { //store the temporary message String publishKey = String.format("%s%d", evt.getClientID(), evt.getMessageID()); - m_inflightStore.put(publishKey, evt); + m_storageService.addInFlight(evt, publishKey); cleanEvt = new CleanInFlightEvent(publishKey); } @@ -293,15 +246,7 @@ public class SimpleMessaging implements IMessaging, Runnable { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it if (!sub.isCleanSession() && !sub.isActive()) { - List storedEvents; - String clientID = evt.getClientID(); - if (!m_persistentMessageStore.containsKey(clientID)) { - storedEvents = new ArrayList(); - } else { - storedEvents = m_persistentMessageStore.get(clientID); - } - storedEvents.add(evt); - m_persistentMessageStore.put(clientID, storedEvents); + m_storageService.storePublishForFuture(evt); } m_notifier.notify(new NotifyEvent(sub.clientId, topic, qos, message, false)); } @@ -313,13 +258,7 @@ public class SimpleMessaging implements IMessaging, Runnable { } if (retain) { - if (message.length == 0) { - //clean the message from topic - m_retainedStore.remove(topic); - } else { - //store the message to the topic - m_retainedStore.put(topic, new StoredMessage(message, qos)); - } + m_storageService.storeRetained(topic, message, qos); } } @@ -327,20 +266,22 @@ public class SimpleMessaging implements IMessaging, Runnable { protected void processSubscribe(SubscribeEvent evt) { LOG.debug("processSubscribe invoked"); Subscription newSubscription = evt.getSubscription(); - String topic = newSubscription.getTopic(); + final String topic = newSubscription.getTopic(); subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription - LOG.debug("Scanning all retained messages, presents are " + m_retainedStore.size()); - for (Map.Entry entry : m_retainedStore) { - StoredMessage storedMsg = entry.getValue(); - if (matchTopics(entry.getKey(), topic)) { - //fire the as retained the message - LOG.debug("Inserting NotifyEvent into outbound for topic " + topic + " entry " + entry.getKey()); - m_notifier.notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(), - storedMsg.getPayload(), true)); + Collection messages = m_storageService.searchMatching(new IMatchingCondition() { + public boolean match(String key) { + return matchTopics(key, topic); //To change body of implemented methods use File | Settings | File Templates. } + }); + + for (StoredMessage storedMsg : messages) { + //fire the as retained the message + LOG.debug("Inserting NotifyEvent into outbound for topic " + topic); + m_notifier.notify(new NotifyEvent(newSubscription.clientId, topic, storedMsg.getQos(), + storedMsg.getPayload(), true)); } } @@ -354,7 +295,7 @@ public class SimpleMessaging implements IMessaging, Runnable { subscriptions.removeForClient(evt.getClientID()); //remove also the messages stored of type QoS1/2 - m_persistentMessageStore.remove(evt.getClientID()); + m_storageService.cleanPersistedPublishes(evt.getClientID()); } private void processDisconnect(DisconnectEvent evt) throws InterruptedException { @@ -367,15 +308,11 @@ public class SimpleMessaging implements IMessaging, Runnable { private void processClose() { LOG.debug("processClose invoked"); - try { - pageFactory.close(); - } catch (IOException ex) { - LOG.error(null, ex); - } + m_storageService.close(); } private void processRepublish(RepublishEvent evt) throws InterruptedException { - List publishedEvents = m_persistentMessageStore.get(evt.getClientID()); + List publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID()); if (publishedEvents == null) { return; } 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 de0ca1ce..ad44d38d 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 @@ -7,6 +7,8 @@ import java.util.Map; import java.util.Queue; import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.LinkedBlockingQueue; + +import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.fusesource.hawtbuf.codec.StringCodec; import org.fusesource.hawtdb.api.BTreeIndexFactory; import org.fusesource.hawtdb.api.MultiIndexFactory; @@ -205,28 +207,23 @@ public class SubscriptionsStore { } private TreeNode subscriptions = new TreeNode(null); private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class); - //persistent Map of clientID, list of Subscriptions - private SortedIndex> m_persistent; + + private IStorageService m_storageService; /** * Initialize basic store structures, like the FS storage to maintain * client's topics subscriptions */ - public void init(MultiIndexFactory multiFactory) { + public void init(IStorageService storageService) { LOG.debug("init invoked"); - BTreeIndexFactory> indexFactory = - new BTreeIndexFactory>(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); - m_persistent = (SortedIndex>) multiFactory.openOrCreate("subscriptions", indexFactory); + m_storageService = storageService; //reload any subscriptions persisted LOG.debug("Reloading all stored subscriptions..."); - for (Map.Entry> entry : m_persistent) { - for (Subscription subscription : entry.getValue()) { - LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic()); - addDirect(subscription); - } + for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) { + LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic()); + addDirect(subscription); } LOG.debug("Finished loading"); } @@ -269,16 +266,11 @@ public class SubscriptionsStore { //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); - m_persistent.put(clientID, subs); + m_storageService.addNewSubscription(newSubscription, clientID); // pageFile.flush(); } - + + public void removeSubscription(String topic, String clientID) { TreeNode matchNode = findMatchingNode(topic); @@ -303,7 +295,7 @@ public class SubscriptionsStore { subscriptions.removeClientSubscriptions(clientID); //remove from log all subscriptions - m_persistent.remove(clientID); + m_storageService.removeAllSubscriptions(clientID); } public void disconnect(String clientID) { diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java new file mode 100644 index 00000000..3be661e1 --- /dev/null +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java @@ -0,0 +1,62 @@ +package org.dna.mqtt.moquette.messaging.spi.impl; + +import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; +import org.dna.mqtt.moquette.messaging.spi.IStorageService; +import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; + +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +/** + */ +public class DummyStorageService implements IStorageService { + public void initStore() { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public Collection searchMatching(IMatchingCondition condition) { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + public void storePublishForFuture(PublishEvent evt) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public List retrivePersistedPublishes(String clientID) { + return null; //To change body of implemented methods use File | Settings | File Templates. + } + + public void cleanPersistedPublishes(String clientID) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void cleanInFlight(String msgID) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void addInFlight(PublishEvent evt, String publishKey) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void addNewSubscription(Subscription newSubscription, String clientID) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void removeAllSubscriptions(String clientID) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public List retrieveAllSubscriptions() { + return Collections.EMPTY_LIST; + } + + public void close() { + //To change body of implemented methods use File | Settings | File Templates. + } +} 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 01a02960..d9039890 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 @@ -40,7 +40,7 @@ public class SubscriptionsStoreTest { pageFactory.open(); PageFile pageFile = pageFactory.getPageFile(); MultiIndexFactory multiIndexFactory = new MultiIndexFactory(pageFile); - store.init(multiIndexFactory); + store.init(new DummyStorageService()); } @After