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..a2454eb2 --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java @@ -0,0 +1,23 @@ +package org.dna.mqtt.moquette.messaging.spi; + +import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; +import org.dna.mqtt.moquette.proto.messages.AbstractMessage; +import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging.StoredMessage; + +import java.util.Collection; + +/** + * 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); + +} 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..14a4654e --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -0,0 +1,72 @@ +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 org.fusesource.hawtbuf.codec.StringCodec; +import org.fusesource.hawtdb.api.BTreeIndexFactory; +import org.fusesource.hawtdb.api.MultiIndexFactory; +import org.fusesource.hawtdb.api.PageFileFactory; +import org.fusesource.hawtdb.api.SortedIndex; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging.StoredMessage; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; +import java.util.Map; + + +/** + * 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; + + private SortedIndex m_retainedStore; + + public HawtDBStorageService(MultiIndexFactory multiIndexFactory) { + m_multiIndexFactory = multiIndexFactory; + } + + + public void initStore() { + BTreeIndexFactory indexFactory = new BTreeIndexFactory(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_retainedStore = (SortedIndex) m_multiIndexFactory.openOrCreate("retained", 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; + } +} 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..c3ac3f53 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 @@ -5,16 +5,20 @@ import java.io.IOException; import java.io.Serializable; import java.text.ParseException; import java.util.ArrayList; +import java.util.Collection; 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; import org.dna.mqtt.moquette.proto.messages.AbstractMessage.QOSType; import org.dna.mqtt.moquette.server.Constants; import org.dna.mqtt.moquette.server.Server; @@ -33,11 +37,12 @@ import org.slf4j.LoggerFactory; */ 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,13 +63,17 @@ public class SimpleMessaging implements IMessaging, Runnable { private BlockingQueue m_inboundQueue = new LinkedBlockingQueue(); private INotifier m_notifier; + + //TODO remove 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; @@ -84,19 +93,14 @@ public class SimpleMessaging implements IMessaging, Runnable { m_multiIndexFactory = new MultiIndexFactory(pageFile); subscriptions.init(m_multiIndexFactory); - initRetainedMessageStore(); initInflightMessageStore(); //init the message store for QoS 1/2 messages in clean sessions initPersistentMessageStore(); - } - - private void initRetainedMessageStore() { - BTreeIndexFactory indexFactory = new BTreeIndexFactory(); - indexFactory.setKeyCodec(StringCodec.INSTANCE); -// m_persistent = indexFactory.openOrCreate(pageFile); - m_retainedStore = (SortedIndex) m_multiIndexFactory.openOrCreate("retained", indexFactory); + m_storageService = new HawtDBStorageService(m_multiIndexFactory); + m_storageService.initStore(); } + /** * Initialize the message store used to handle the temporary storage of QoS 1,2 @@ -313,13 +317,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 +325,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)); } }