Extracted the persistence dependency from SubscriptionsStore

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

@@ -25,9 +25,15 @@ import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
  */
 public interface IPersistentSubscriptionStore {
 
+    /**
+     * Add a new subscription to the session
+     */
     void addNewSubscription(Subscription newSubscription, String clientID);
 
-    void removeAllSubscriptions(String clientID);
+    /**
+     * Remove all the subscriptions of the session
+     */
+    void wipeSubscriptions(String sessionID);
 
     List<Subscription> retrieveAllSubscriptions();
 }

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

@@ -260,7 +260,7 @@ public class HawtDBStorageService implements IStorageService {
         }
     }
 
-    public void removeAllSubscriptions(String clientID) {
+    public void wipeSubscriptions(String clientID) {
         m_persistentSubscriptions.remove(clientID);
     }
 

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

@@ -130,7 +130,7 @@ public class MemoryStorageService implements IStorageService {
         }
     }
 
-    public void removeAllSubscriptions(String clientID) {
+    public void wipeSubscriptions(String clientID) {
         m_persistentSubscriptions.remove(clientID);
     }
 

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

@@ -160,7 +160,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
             boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION);
             if (cleanSession) {
                 //cleanup topic subscriptions
-                processRemoveAllSubscriptions(msg.getClientID());
+                cleanSession(msg.getClientID());
             }
 
             m_clientIDs.get(msg.getClientID()).getSession().close(false);
@@ -208,7 +208,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         if (msg.isCleanSession()) {
             //remove all prev subscriptions
             //cleanup topic subscriptions
-            processRemoveAllSubscriptions(msg.getClientID());
+            cleanSession(msg.getClientID());
         }
 
         ConnAckMessage okResp = new ConnAckMessage();
@@ -243,10 +243,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         m_storageService.cleanPersistedPublishMessage(clientID, messageID);
     }
     
-    private void processRemoveAllSubscriptions(String clientID) {
+    private void cleanSession(String clientID) {
         LOG.info("cleaning old saved subscriptions for client <{}>", clientID);
         //remove from log all subscriptions
-        m_storageService.removeAllSubscriptions(clientID);
+        m_storageService.wipeSubscriptions(clientID);
         subscriptions.removeForClient(clientID);
 
         //remove also the messages stored of type QoS1/2
@@ -458,7 +458,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     void processDisconnect(ServerChannel session, String clientID, boolean cleanSession) throws InterruptedException {
         if (cleanSession) {
             //cleanup topic subscriptions
-            processRemoveAllSubscriptions(clientID);
+            cleanSession(clientID);
         }
 //        m_notifier.disconnect(evt.getSession());
         m_clientIDs.remove(clientID);