Extracted the persistence dependency from SubscriptionsStore

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java

@@ -245,6 +245,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     
     private void processRemoveAllSubscriptions(String clientID) {
         LOG.info("cleaning old saved subscriptions for client <{}>", clientID);
+        //remove from log all subscriptions
+        m_storageService.removeAllSubscriptions(clientID);
         subscriptions.removeForClient(clientID);
 
         //remove also the messages stored of type QoS1/2
@@ -534,6 +536,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         LOG.info("<{}> subscribed to topic <{}> with QoS {}", 
                 newSubscription.getClientId(), topic, 
                 AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos()));
+        String clientID = newSubscription.getClientId();
+        m_storageService.addNewSubscription(newSubscription, clientID);
         subscriptions.add(newSubscription);
 
         //scans retained messages to be published to the new subscription

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

@@ -19,6 +19,7 @@ import com.lmax.disruptor.BatchEventProcessor;
 import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.RingBuffer;
 import com.lmax.disruptor.SequenceBarrier;
+import java.util.List;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -27,6 +28,7 @@ import java.util.concurrent.TimeUnit;
 import org.dna.mqtt.moquette.messaging.spi.IMessaging;
 import org.dna.mqtt.moquette.messaging.spi.IStorageService;
 import org.dna.mqtt.moquette.messaging.spi.impl.events.*;
+import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
 import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore;
 import org.dna.mqtt.moquette.proto.messages.PubCompMessage;
 import org.dna.mqtt.moquette.proto.messages.*;
@@ -40,7 +42,9 @@ import org.slf4j.LoggerFactory;
  *
  * Singleton class that orchestrate the execution of the protocol.
  *
- * Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion.
+ * Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion;
+ * the requests income from front Netty connectors and are dispatched to the 
+ * ProtocolProcessor.
  *
  * @author andrea
  */
@@ -199,8 +203,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     private void processInit(Properties props) {
         m_storageService = new HawtDBStorageService();
         m_storageService.initStore();
-
-        subscriptions.init(m_storageService);
+        List<Subscription> storedSubscriptions = m_storageService.retrieveAllSubscriptions();
+        subscriptions.init(storedSubscriptions);
         
         String passwdPath = props.getProperty("password_file");
         String configPath = System.getProperty("moquette.path", null);

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java

@@ -21,7 +21,6 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.LinkedBlockingDeque;
-import org.dna.mqtt.moquette.messaging.spi.IPersistentSubscriptionStore;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -73,23 +72,18 @@ public class SubscriptionsStore {
     private TreeNode subscriptions = new TreeNode(null);
     private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
 
-    private IPersistentSubscriptionStore m_storageService;
-
     /**
-     * Initialize basic store structures, like the FS storage to maintain
-     * client's topics subscriptions
+     * Initialize the subscription tree with the list of subscriptions.
      */
-    public void init(IPersistentSubscriptionStore storageService) {
+    public void init(List<Subscription> subscriptions) {
         LOG.debug("init invoked");
 
-        m_storageService = storageService;
-
         //reload any subscriptions persisted
         if (LOG.isDebugEnabled()) {
             LOG.debug("Reloading all stored subscriptions...subscription tree before {}", dumpTree());
         }
         
-        for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) {
+        for (Subscription subscription : subscriptions) {
             LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopicFilter());
             addDirect(subscription);
         }
@@ -135,8 +129,8 @@ public class SubscriptionsStore {
         addDirect(newSubscription);
 
         //log the subscription
-        String clientID = newSubscription.getClientId();
-        m_storageService.addNewSubscription(newSubscription, clientID);
+//        String clientID = newSubscription.getClientId();
+//        m_storageService.addNewSubscription(newSubscription, clientID);
     }
 
 
@@ -177,7 +171,7 @@ public class SubscriptionsStore {
         subscriptions.removeClientSubscriptions(clientID);
 
         //remove from log all subscriptions
-        m_storageService.removeAllSubscriptions(clientID);
+//        m_storageService.removeAllSubscriptions(clientID);
     }
 
     public void deactivate(String clientID) {