Fixed bug #400 a transient session (clean=true) that tried to lookup session from persistent store

broker/src/main/java/io/moquette/spi/ClientSession.java

@@ -115,4 +115,6 @@ public abstract class ClientSession {
     public String toString() {
         return getClass().getSimpleName() + "{clientID='" + clientID + "'}";
     }
+
+    public abstract void wipeSubscriptions();
 }

broker/src/main/java/io/moquette/spi/DurableSession.java

@@ -210,4 +210,9 @@ public class DurableSession extends ClientSession {
     public int getInflightMessagesNo() {
         return this.sessionsStore.getInflightMessagesNo(clientID);
     }
+
+    @Override
+    public void wipeSubscriptions() {
+        this.subscriptionsStore.wipeSubscriptions(clientID);
+    }
 }

broker/src/main/java/io/moquette/spi/TransientSession.java

@@ -182,4 +182,9 @@ public class TransientSession extends ClientSession {
     public int getInflightMessagesNo() {
         return this.inboundInflightMap.size();
     }
+
+    @Override
+    public void wipeSubscriptions() {
+        // it's transient, no need to wide subs because wasn't stored
+    }
 }

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java

@@ -235,7 +235,7 @@ public class ProtocolProcessor {
         }
 
         ConnectionDescriptor descriptor = new ConnectionDescriptor(clientId, channel, cleanSession);
-        ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor);
+        final ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor);
         if (existing != null) {
             LOG.info("Client ID is being used in an existing connection, force to be closed. CId={}", clientId);
             existing.abort();
@@ -616,7 +616,8 @@ public class ProtocolProcessor {
 
         if (descriptor.cleanSession) {
             LOG.info("Removing saved subscriptions. CId={}", descriptor.clientID);
-            subscriptionStore.wipeSubscriptions(clientID);
+            final ClientSession session = this.sessionsRepository.sessionForClient(clientID);
+            session.wipeSubscriptions();
             LOG.info("Saved subscriptions have been removed. CId={}", descriptor.clientID);
         }
         return true;

broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java

@@ -110,11 +110,6 @@ public class SessionsRepository {
         return result;
     }
 
-    private void updateCleanStatus(String clientID, boolean cleanSession) {
-        LOG.info("Updating cleanSession flag. CId={}, cleanSession={}", clientID, cleanSession);
-        this.sessions.updateCleanStatus(clientID, cleanSession);
-    }
-
     ClientSession createOrLoadClientSession(String clientId, boolean cleanSession) {
         ClientSession clientSession = this.sessionForClient(clientId);
         if (clientSession == null) {