From 665e60a646f280c24c1a04167124c8efa40fa5dc Mon Sep 17 00:00:00 2001
From: "selva.andre" <selva.andre@gmail.com>
Date: Mon, 30 Jun 2014 22:59:37 +0200
Subject: [PATCH] Extracted the persistence dependency from SubscriptionsStore

---
 .../messaging/spi/IPersistentSubscriptionStore.java    |  8 +++++++-
 .../messaging/spi/impl/HawtDBStorageService.java       |  2 +-
 .../messaging/spi/impl/MemoryStorageService.java       |  2 +-
 .../moquette/messaging/spi/impl/ProtocolProcessor.java | 10 +++++-----
 4 files changed, 14 insertions(+), 8 deletions(-)

diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java
index 17404dc4..49a32487 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IPersistentSubscriptionStore.java
+++ b/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();
 }
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
index ce8e0900..979180fd 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java
+++ b/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);
     }
 
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
index 13b5d902..06e0af64 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/MemoryStorageService.java
+++ b/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);
     }
 
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
index 81db4d18..8b072410 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+++ b/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);