Moved Subscriptions storage from SubscriptionStore to StorageService

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<Subscription> retrieveAllSubscriptions();
 }

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<String, PublishEvent> m_inflightStore;
 
+    //persistent Map of clientID, list of Subscriptions
+    //TODO rename into persistentSubscriptions
+    private SortedIndex<String, List<Subscription>> 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<String, List<PublishEvent>>) m_multiIndexFactory.openOrCreate("persistedMessages", indexFactory);
     }
 
+    private void initPersistentSubscriptions() {
+        BTreeIndexFactory<String, List<Subscription>> indexFactory =
+                new BTreeIndexFactory<String, List<Subscription>>();
+        indexFactory.setKeyCodec(StringCodec.INSTANCE);
+
+        m_persistent = (SortedIndex<String, List<Subscription>>) 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<Subscription>());
+        }
+
+        List<Subscription> subs = m_persistent.get(clientID);
+        subs.add(newSubscription);
+        m_persistent.put(clientID, subs);
+    }
+
+    public void removeAllSubscriptions(String clientID) {
+        m_persistent.remove(clientID);
+    }
+
+    public List<Subscription> retrieveAllSubscriptions() {
+        List<Subscription> allSubscriptions = new ArrayList<Subscription>();
+        for (Map.Entry<String, List<Subscription>> entry : m_persistent) {
+            allSubscriptions.addAll(entry.getValue());
+        }
+        return allSubscriptions;
+    }
 }

parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java

@@ -82,10 +82,11 @@ public class SimpleMessaging implements IMessaging, Runnable {
         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);
     }
 
     

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<String, List<Subscription>> 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<String, List<Subscription>> indexFactory =
-                new BTreeIndexFactory<String, List<Subscription>>();
-        indexFactory.setKeyCodec(StringCodec.INSTANCE);
 
-        m_persistent = (SortedIndex<String, List<Subscription>>) multiFactory.openOrCreate("subscriptions", indexFactory);
+        m_storageService = storageService;
 
         //reload any subscriptions persisted
         LOG.debug("Reloading all stored subscriptions...");
-        for (Map.Entry<String, List<Subscription>> entry : m_persistent) {
+        for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) {
-            for (Subscription subscription : entry.getValue()) {
+            LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic());
-                LOG.debug("Re-subscribing " + subscription.getClientId() + " to topic " + subscription.getTopic());
+            addDirect(subscription);
-                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_storageService.addNewSubscription(newSubscription, clientID);
-            m_persistent.put(clientID, new ArrayList<Subscription>());
-        }
-
-        List<Subscription> subs = m_persistent.get(clientID);
-        subs.add(newSubscription);
-        m_persistent.put(clientID, subs);
 //        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) {

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<SimpleMessaging.StoredMessage> 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<PublishEvent> 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<Subscription> retrieveAllSubscriptions() {
+        return Collections.EMPTY_LIST;
+    }
+}

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