From 5456431dd4c625e3be126ba423244a375efab65f Mon Sep 17 00:00:00 2001
From: andsel <selva.andre@gmail.com>
Date: Fri, 27 Apr 2018 12:44:48 +0200
Subject: [PATCH] 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      | 2 ++
 broker/src/main/java/io/moquette/spi/DurableSession.java     | 5 +++++
 broker/src/main/java/io/moquette/spi/TransientSession.java   | 5 +++++
 .../main/java/io/moquette/spi/impl/ProtocolProcessor.java    | 5 +++--
 .../main/java/io/moquette/spi/impl/SessionsRepository.java   | 5 -----
 5 files changed, 15 insertions(+), 7 deletions(-)

diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java
index 042110fa..5de51f16 100644
--- a/broker/src/main/java/io/moquette/spi/ClientSession.java
+++ b/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();
 }
diff --git a/broker/src/main/java/io/moquette/spi/DurableSession.java b/broker/src/main/java/io/moquette/spi/DurableSession.java
index 60d8b74e..b8aa09e4 100644
--- a/broker/src/main/java/io/moquette/spi/DurableSession.java
+++ b/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);
+    }
 }
diff --git a/broker/src/main/java/io/moquette/spi/TransientSession.java b/broker/src/main/java/io/moquette/spi/TransientSession.java
index de83035e..288ae9d6 100644
--- a/broker/src/main/java/io/moquette/spi/TransientSession.java
+++ b/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
+    }
 }
diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
index cc5e76d1..035ea017 100644
--- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java
+++ b/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;
diff --git a/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java b/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java
index 5a7666aa..a3bfd7e9 100644
--- a/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java
+++ b/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) {