Extracted persistence interface for subscriptions from IStorageService

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

@@ -0,0 +1,18 @@
+package org.dna.mqtt.moquette.messaging.spi;
+
+import java.util.List;
+import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
+
+/**
+ * Store used to handle the persistence of the subscriptions tree.
+ *
+ * @author andrea
+ */
+public interface IPersistentSubscriptionStore {
+
+    void addNewSubscription(Subscription newSubscription, String clientID);
+
+    void removeAllSubscriptions(String clientID);
+
+    List<Subscription> retrieveAllSubscriptions();
+}

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

@@ -1,7 +1,6 @@
 package org.dna.mqtt.moquette.messaging.spi;
 
 import java.nio.ByteBuffer;
-import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.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.HawtDBStorageService.StoredMessage;
@@ -12,7 +11,7 @@ import java.util.List;
 /**
  * Defines the SPI to be implemented by a StorageService that handle persistence of messages and subscriptions.
  */
-public interface IStorageService {
+public interface IStorageService extends IPersistentSubscriptionStore {
 
     /**
      * Used to initialize all persistent store structures
@@ -35,12 +34,6 @@ public interface IStorageService {
 
     void addInFlight(PublishEvent evt, String publishKey);
 
-    void addNewSubscription(Subscription newSubscription, String clientID);
-
-    void removeAllSubscriptions(String clientID);
-
-    List<Subscription> retrieveAllSubscriptions();
-
     void close();
 
     void persistQoS2Message(String publishKey, PublishEvent evt);

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

@@ -60,8 +60,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
 
     ProtocolProcessor() {}
     
-    void init(/*Map<String, ConnectionDescriptor> clientIDs, */SubscriptionsStore subscriptions,
-            IStorageService storageService) {
+    /**
+     * @param subscriptions the subscription store where are stored all the existing
+     *  clients subscriptions.
+     * @param storageService the persistent store to use for save/load of messages
+     *  for QoS1 and QoS2 handling.
+     */
+    void init(SubscriptionsStore subscriptions, IStorageService storageService) {
         //m_clientIDs = clientIDs;
         this.subscriptions = subscriptions;
         m_storageService = storageService;

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

@@ -6,8 +6,8 @@ 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.dna.mqtt.moquette.messaging.spi.IStorageService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -58,13 +58,13 @@ public class SubscriptionsStore {
     private TreeNode subscriptions = new TreeNode(null);
     private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class);
 
-    private IStorageService m_storageService;
+    private IPersistentSubscriptionStore m_storageService;
 
     /**
      * Initialize basic store structures, like the FS storage to maintain
      * client's topics subscriptions
      */
-    public void init(IStorageService storageService) {
+    public void init(IPersistentSubscriptionStore storageService) {
         LOG.debug("init invoked");
 
         m_storageService = storageService;