From 2f9c957310c3c6b3eb2fb4dab944db290a3fe9e9 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 16:02:48 +0200 Subject: [PATCH 1/7] Moved retained storage part into StorageService --- .../messaging/spi/IMatchingCondition.java | 8 +++ .../messaging/spi/IStorageService.java | 23 ++++++ .../spi/impl/HawtDBStorageService.java | 72 +++++++++++++++++++ .../messaging/spi/impl/SimpleMessaging.java | 54 +++++++------- 4 files changed, 130 insertions(+), 27 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMatchingCondition.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IStorageService.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java 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)); } } From 2ed87b2c1a52ac8e86011577faf2078bd34c9df6 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 16:18:09 +0200 Subject: [PATCH 2/7] Moved the part to manage Pusblish messages persistence to StorageService --- .../messaging/spi/IStorageService.java | 8 ++++ .../spi/impl/HawtDBStorageService.java | 38 +++++++++++++++++++ .../messaging/spi/impl/SimpleMessaging.java | 25 ++---------- 3 files changed, 49 insertions(+), 22 deletions(-) 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 a2454eb2..6f0c9132 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,10 +1,12 @@ package org.dna.mqtt.moquette.messaging.spi; import org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging; +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. @@ -20,4 +22,10 @@ public interface IStorageService { Collection searchMatching(IMatchingCondition condition); + void storePublishForFuture(PublishEvent evt); + + List retrivePersistedPublishes(String clientID); + + void cleanPersistedPublishes(String clientID); + } 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 14a4654e..6491d64b 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 @@ -3,6 +3,8 @@ 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.messaging.spi.impl.events.RemoveAllSubscriptionsEvent; +import org.dna.mqtt.moquette.messaging.spi.impl.events.RepublishEvent; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.fusesource.hawtbuf.codec.StringCodec; import org.fusesource.hawtdb.api.BTreeIndexFactory; @@ -30,6 +32,8 @@ public class HawtDBStorageService implements IStorageService { private MultiIndexFactory m_multiIndexFactory; private PageFileFactory pageFactory; + //maps clientID to the list of pending messages stored + private SortedIndex> m_persistentMessageStore; private SortedIndex m_retainedStore; public HawtDBStorageService(MultiIndexFactory multiIndexFactory) { @@ -38,6 +42,12 @@ public class HawtDBStorageService implements IStorageService { public void initStore() { + initRetainedStore(); + //init the message store for QoS 1/2 messages in clean sessions + initPersistentMessageStore(); + } + + private void initRetainedStore() { BTreeIndexFactory indexFactory = new BTreeIndexFactory(); indexFactory.setKeyCodec(StringCodec.INSTANCE); @@ -45,6 +55,14 @@ public class HawtDBStorageService implements IStorageService { } + + private void initPersistentMessageStore() { + BTreeIndexFactory> indexFactory = new BTreeIndexFactory>(); + indexFactory.setKeyCodec(StringCodec.INSTANCE); + + m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory); + } + public void storeRetained(String topic, byte[] message, AbstractMessage.QOSType qos) { if (message.length == 0) { //clean the message from topic @@ -69,4 +87,24 @@ public class HawtDBStorageService implements IStorageService { 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); + } } 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 c3ac3f53..2602dbf0 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 @@ -18,7 +18,6 @@ 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; @@ -70,8 +69,6 @@ public class SimpleMessaging implements IMessaging, Runnable { //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; @@ -94,8 +91,6 @@ public class SimpleMessaging implements IMessaging, Runnable { subscriptions.init(m_multiIndexFactory); initInflightMessageStore(); - //init the message store for QoS 1/2 messages in clean sessions - initPersistentMessageStore(); m_storageService = new HawtDBStorageService(m_multiIndexFactory); m_storageService.initStore(); @@ -113,12 +108,6 @@ public class SimpleMessaging implements IMessaging, Runnable { 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; @@ -297,15 +286,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)); } @@ -354,7 +335,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 { @@ -375,7 +356,7 @@ public class SimpleMessaging implements IMessaging, Runnable { } 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; } From b092d15c12c2de6ad01a0edffe75abade5e36b03 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 16:28:03 +0200 Subject: [PATCH 3/7] Moved the fling messages part persistence into StorageService --- .../messaging/spi/IStorageService.java | 3 ++ .../spi/impl/HawtDBStorageService.java | 22 ++++++++++ .../messaging/spi/impl/SimpleMessaging.java | 42 +++++-------------- 3 files changed, 36 insertions(+), 31 deletions(-) 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 6f0c9132..fb0d43c6 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 @@ -28,4 +28,7 @@ public interface IStorageService { void cleanPersistedPublishes(String clientID); + void cleanInFlight(String msgID); + + void addInFlight(PublishEvent evt, String publishKey); } 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 6491d64b..27e592dc 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 @@ -2,6 +2,7 @@ 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.CleanInFlightEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.RemoveAllSubscriptionsEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.RepublishEvent; @@ -35,6 +36,8 @@ public class HawtDBStorageService implements IStorageService { //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; public HawtDBStorageService(MultiIndexFactory multiIndexFactory) { m_multiIndexFactory = multiIndexFactory; @@ -63,6 +66,17 @@ public class HawtDBStorageService implements IStorageService { m_persistentMessageStore = (SortedIndex>) m_multiIndexFactory.openOrCreate("persistedMessages", 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 @@ -107,4 +121,12 @@ public class HawtDBStorageService implements IStorageService { 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); + } } 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 2602dbf0..da0108fc 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,15 +1,5 @@ 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.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; @@ -21,15 +11,21 @@ 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 @@ -67,9 +63,6 @@ public class SimpleMessaging implements IMessaging, Runnable { private MultiIndexFactory m_multiIndexFactory; private PageFileFactory pageFactory; - //bind clientID+MsgID -> evt message published - private SortedIndex m_inflightStore; - private IStorageService m_storageService; public SimpleMessaging() { @@ -90,25 +83,12 @@ public class SimpleMessaging implements IMessaging, Runnable { m_multiIndexFactory = new MultiIndexFactory(pageFile); subscriptions.init(m_multiIndexFactory); - initInflightMessageStore(); m_storageService = new HawtDBStorageService(m_multiIndexFactory); m_storageService.initStore(); } - /** - * 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 setNotifier(INotifier notifier) { m_notifier= notifier; } @@ -251,7 +231,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); } @@ -274,7 +254,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); } From 6f88060ffe3ff43329bf0c66282011ceabb7d917 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 16:44:47 +0200 Subject: [PATCH 4/7] Missed initialization in previous commit --- .../mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java | 1 + 1 file changed, 1 insertion(+) 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 27e592dc..4f9d52d2 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 @@ -48,6 +48,7 @@ public class HawtDBStorageService implements IStorageService { initRetainedStore(); //init the message store for QoS 1/2 messages in clean sessions initPersistentMessageStore(); + initInflightMessageStore(); } private void initRetainedStore() { From ae8a2afb8b0a65e9fc205d96a4fbdeeafe545633 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 16:55:19 +0200 Subject: [PATCH 5/7] 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 From a5d28b67e90a4c22c2cf7a4f57d34977dc295d62 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 17:00:22 +0200 Subject: [PATCH 6/7] Moved all HawtDB stuff into HawtDBStorageService --- .../messaging/spi/IStorageService.java | 2 + .../spi/impl/HawtDBStorageService.java | 42 ++++++++++++++----- .../messaging/spi/impl/SimpleMessaging.java | 29 +------------ .../spi/impl/DummyStorageService.java | 4 ++ 4 files changed, 39 insertions(+), 38 deletions(-) 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 b6633a84..0ee6b843 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 @@ -38,4 +38,6 @@ public interface IStorageService { 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 index da85628a..76736f97 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 @@ -1,27 +1,25 @@ 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.CleanInFlightEvent; import org.dna.mqtt.moquette.messaging.spi.impl.events.PublishEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.RemoveAllSubscriptionsEvent; -import org.dna.mqtt.moquette.messaging.spi.impl.events.RepublishEvent; 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.BTreeIndexFactory; -import org.fusesource.hawtdb.api.MultiIndexFactory; -import org.fusesource.hawtdb.api.PageFileFactory; -import org.fusesource.hawtdb.api.SortedIndex; +import org.fusesource.hawtdb.api.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import static org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging.StoredMessage; - +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 @@ -43,8 +41,22 @@ public class HawtDBStorageService implements IStorageService { //TODO rename into persistentSubscriptions private SortedIndex> m_persistent; - public HawtDBStorageService(MultiIndexFactory multiIndexFactory) { - m_multiIndexFactory = multiIndexFactory; + 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); } @@ -165,4 +177,12 @@ public class HawtDBStorageService implements IStorageService { } 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 7bf1588c..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 @@ -59,31 +59,10 @@ public class SimpleMessaging implements IMessaging, Runnable { private INotifier m_notifier; - //TODO remove - private MultiIndexFactory m_multiIndexFactory; - private PageFileFactory pageFactory; - 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); - - - m_storageService = new HawtDBStorageService(m_multiIndexFactory); + m_storageService = new HawtDBStorageService(); m_storageService.initStore(); subscriptions.init(m_storageService); @@ -329,11 +308,7 @@ 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 { 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 index 594f4ab4..3be661e1 100644 --- 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 @@ -55,4 +55,8 @@ public class DummyStorageService implements IStorageService { public List retrieveAllSubscriptions() { return Collections.EMPTY_LIST; } + + public void close() { + //To change body of implemented methods use File | Settings | File Templates. + } } From 8519b47dc2a5f0bb6c873581e51139d2aab63162 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Sat, 1 Sep 2012 17:00:57 +0200 Subject: [PATCH 7/7] Minor change, field renaming --- .../spi/impl/HawtDBStorageService.java | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) 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 76736f97..85848203 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 @@ -38,8 +38,7 @@ public class HawtDBStorageService implements IStorageService { private SortedIndex m_inflightStore; //persistent Map of clientID, list of Subscriptions - //TODO rename into persistentSubscriptions - private SortedIndex> m_persistent; + private SortedIndex> m_persistentSubscriptions; public HawtDBStorageService() { String storeFile = Server.STORAGE_FILE_PATH; @@ -89,7 +88,7 @@ public class HawtDBStorageService implements IStorageService { new BTreeIndexFactory>(); indexFactory.setKeyCodec(StringCodec.INSTANCE); - m_persistent = (SortedIndex>) m_multiIndexFactory.openOrCreate("subscriptions", indexFactory); + m_persistentSubscriptions = (SortedIndex>) m_multiIndexFactory.openOrCreate("subscriptions", indexFactory); } /** @@ -157,22 +156,22 @@ public class HawtDBStorageService implements IStorageService { } public void addNewSubscription(Subscription newSubscription, String clientID) { - if (!m_persistent.containsKey(clientID)) { - m_persistent.put(clientID, new ArrayList()); + if (!m_persistentSubscriptions.containsKey(clientID)) { + m_persistentSubscriptions.put(clientID, new ArrayList()); } - List subs = m_persistent.get(clientID); + List subs = m_persistentSubscriptions.get(clientID); subs.add(newSubscription); - m_persistent.put(clientID, subs); + m_persistentSubscriptions.put(clientID, subs); } public void removeAllSubscriptions(String clientID) { - m_persistent.remove(clientID); + m_persistentSubscriptions.remove(clientID); } public List retrieveAllSubscriptions() { List allSubscriptions = new ArrayList(); - for (Map.Entry> entry : m_persistent) { + for (Map.Entry> entry : m_persistentSubscriptions) { allSubscriptions.addAll(entry.getValue()); } return allSubscriptions;