Fixed missed clean session for ClientSession, issue #145

ChangeLog.txt

@@ -1,7 +1,8 @@
 Version 0.9:
     Moved canRead check to topic from publish phase to subscribe (#144)
     Added javadoc and src jars creations for JCenter/Maven central
-    
+    Added persisting of clean session flag for already existing client session (fix issue #145)
+
 Version 0.8:
     Introduced authenticator and authorizator implementations.
     Fixed bug, doesn't close the channel when connect doesn't contains a valid proto name, issue #65

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

@@ -55,7 +55,7 @@ public class ClientSession {
 
     private Set<Subscription> subscriptions = new HashSet<>();
 
-    public final boolean cleanSession;
+    private volatile boolean cleanSession;
 
     private boolean active = false;
 
@@ -146,6 +146,11 @@ public class ClientSession {
         return this.cleanSession;
     }
 
+    public void cleanSession(boolean cleanSession) {
+        this.cleanSession = cleanSession;
+        this.m_sessionsStore.updateCleanStatus(this.clientID, cleanSession);
+    }
+
     public void activate() {
         this.active = true;
     }

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

@@ -27,6 +27,8 @@ import io.moquette.spi.impl.subscriptions.Subscription;
  */
 public interface ISessionsStore {
 
+    void updateCleanStatus(String clientID, boolean cleanSession);
+
     class ClientTopicCouple {
         public final String topicFilter;
         public final String clientID;
@@ -99,6 +101,7 @@ public interface ISessionsStore {
     ClientSession createNewSession(String clientID, boolean cleanSession);
 
     /**
+     * @param clientID the client owning the session.
      * @return the session for the given clientID, null if not found.
      * */
     ClientSession sessionForClient(String clientID);

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

@@ -219,6 +219,9 @@ public class ProtocolProcessor {
         if (!msg.isCleanSession() && isSessionAlreadyStored) {
             okResp.setSessionPresent(true);
         }
+        if (isSessionAlreadyStored) {
+            clientSession.cleanSession(msg.isCleanSession());
+        }
         channel.writeAndFlush(okResp);
         m_interceptor.notifyClientConnected(msg);
 

broker/src/main/java/io/moquette/spi/persistence/MapDBSessionsStore.java

@@ -143,6 +143,11 @@ class MapDBSessionsStore implements ISessionsStore {
         return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
     }
 
+    @Override
+    public void updateCleanStatus(String clientID, boolean cleanSession) {
+        m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
+    }
+
     /**
      * Return the next valid packetIdentifier for the given client session.
      * */

broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java

@@ -19,6 +19,7 @@ import io.moquette.proto.messages.AbstractMessage;
 import io.moquette.proto.messages.ConnectMessage;
 import io.moquette.proto.messages.SubscribeMessage;
 import io.moquette.server.netty.NettyUtils;
+import io.moquette.spi.ClientSession;
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.impl.subscriptions.SubscriptionsStore;
@@ -207,6 +208,35 @@ public class ProtocolProcessor_CONNECT_Test {
         assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
     }
 
+    @Test
+    public void connectWithCleanSessionUpdateClientSession() throws InterruptedException {
+        m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true,
+                new PermitAllAuthorizator(), ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
+
+        //first connect with clean session true
+        connMsg.setClientID("123");
+        connMsg.setCleanSession(true);
+        m_processor.processConnect(m_session, connMsg);
+        assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
+        m_processor.processDisconnect(m_session);
+        assertFalse(m_session.isOpen());
+
+        //second connect with clean session false
+        m_session = new EmbeddedChannel();
+        ConnectMessage secondConnMsg = new ConnectMessage();
+        secondConnMsg.setProtocolVersion((byte) 0x03);
+        secondConnMsg.setClientID("123");
+        m_processor.processConnect(m_session, secondConnMsg);
+        assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
+
+        //Verify client session is clean false
+        ClientSession session = m_sessionStore.sessionForClient("123");
+        assertFalse(session.isCleanSession());
+
+        //Verify
+        //assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
+    }
+
     @Test
     public void connectWithSameClientIDBadCredentialsDoesntDropExistingClient() {
         //Connect a client1

broker/src/test/java/io/moquette/spi/persistence/MemorySessionStore.java

@@ -125,6 +125,11 @@ public class MemorySessionStore implements ISessionsStore {
         return new ClientSession(clientID, m_messagesStore, this, storedSession.cleanSession);
     }
 
+    @Override
+    public void updateCleanStatus(String clientID, boolean cleanSession) {
+        m_persistentSessions.put(clientID, new MapDBPersistentStore.PersistentSession(cleanSession));
+    }
+
     @Override
     public List<ClientTopicCouple> listAllSubscriptions() {
         List<ClientTopicCouple> allSubscriptions = new ArrayList<>();