From ae8a2afb8b0a65e9fc205d96a4fbdeeafe545633 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 16:55:19 +0200 Subject: [PATCH] Moved Subscriptions storage from SubscriptionStore to StorageService --- .../messaging/spi/IStorageService.java | 7 +++ .../spi/impl/HawtDBStorageService.java | 35 +++++++++++ .../messaging/spi/impl/SimpleMessaging.java | 5 +- .../spi/impl/SubscriptionsStore.java | 34 +++++------ .../spi/impl/DummyStorageService.java | 58 +++++++++++++++++++ .../spi/impl/SubscriptionsStoreTest.java | 2 +- 6 files changed, 117 insertions(+), 24 deletions(-) create mode 100644 parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java 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 index fb0d43c6..b6633a84 100644 --- 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 @@ -1,6 +1,7 @@ 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; @@ -31,4 +32,10 @@ public interface IStorageService { void cleanInFlight(String msgID); void addInFlight(PublishEvent evt, String publishKey); + + void addNewSubscription(Subscription newSubscription, String clientID); + + void removeAllSubscriptions(String clientID); + + List retrieveAllSubscriptions(); } 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 index 4f9d52d2..da85628a 100644 --- 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 @@ -39,6 +39,10 @@ public class HawtDBStorageService implements IStorageService { //bind clientID+MsgID -> evt message published private SortedIndex m_inflightStore; + //persistent Map of clientID, list of Subscriptions + //TODO rename into persistentSubscriptions + private SortedIndex> m_persistent; + public HawtDBStorageService(MultiIndexFactory multiIndexFactory) { m_multiIndexFactory = multiIndexFactory; } @@ -49,6 +53,7 @@ public class HawtDBStorageService implements IStorageService { //init the message store for QoS 1/2 messages in clean sessions initPersistentMessageStore(); initInflightMessageStore(); + initPersistentSubscriptions(); } private void initRetainedStore() { @@ -67,6 +72,14 @@ public class HawtDBStorageService implements IStorageService { m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); } + private void initPersistentSubscriptions() { + BTreeIndexFactory> indexFactory = + new BTreeIndexFactory>(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_persistent = (SortedIndex>) m_multiIndexFactory.openOrCreate("subscriptions", indexFactory); + } + /** * Initialize the message store used to handle the temporary storage of QoS 1,2 * messages in flight. @@ -130,4 +143,26 @@ public class HawtDBStorageService implements IStorageService { public void addInFlight(PublishEvent evt, String publishKey) { m_inflightStore.put(publishKey, evt); } + + public void addNewSubscription(Subscription newSubscription, String clientID) { + 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); + } + + public void removeAllSubscriptions(String clientID) { + m_persistent.remove(clientID); + } + + public List retrieveAllSubscriptions() { + List allSubscriptions = new ArrayList(); + for (Map.Entry> entry : m_persistent) { + allSubscriptions.addAll(entry.getValue()); + } + return allSubscriptions; + } } 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 da0108fc..7bf1588c 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 @@ -81,11 +81,12 @@ public class SimpleMessaging implements IMessaging, Runnable { pageFactory.open(); PageFile pageFile = pageFactory.getPageFile(); m_multiIndexFactory = new MultiIndexFactory(pageFile); - - subscriptions.init(m_multiIndexFactory); + m_storageService = new HawtDBStorageService(m_multiIndexFactory); m_storageService.initStore(); + + subscriptions.init(m_storageService); } 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..594f4ab4 --- /dev/null +++ b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/DummyStorageService.java @@ -0,0 +1,58 @@ +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; + } +} 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