用户退出时，多加一种选择，停用session不清除session

broker/src/main/java/io/moquette/persistence/MemorySessionStore.java

@@ -672,6 +672,18 @@ public class MemorySessionStore implements ISessionsStore {
         return ErrorCode.ERROR_CODE_SUCCESS;
     }
 
+
+    @Override
+    public void disableSession(String clientId) {
+        Session session = sessions.get(clientId);
+        if (session != null && session.getDeleted() == 0) {
+            databaseStore.updateSessionDeleted(session.getUsername(), clientId, 1);
+            ConcurrentSkipListSet<String> sessionSet = getUserSessionSet(session.username);
+            sessionSet.remove(clientId);
+            sessions.remove(clientId);
+        }
+    }
+
     @Override
     public void cleanSession(String clientID) {
         LOG.info("Fooooooooo <{}>", clientID);

broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java

@@ -70,7 +70,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
                     m_processor.processPubRel(ctx.channel(), msg);
                     break;
                 case DISCONNECT:
-                    m_processor.processDisconnect(ctx.channel(), msg.fixedHeader().isDup());
+                    m_processor.processDisconnect(ctx.channel(), msg.fixedHeader().isDup(), msg.fixedHeader().isRetain());
                     break;
                 case PUBACK:
                     m_processor.processPubAck(ctx.channel(), (MqttPubAckMessage) msg);

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

@@ -109,11 +109,14 @@ public class ClientSession {
     }
 
 
-    public void disconnect(boolean cleanSession) {
+    public void disconnect(boolean cleanSession, boolean disableSession) {
         if (cleanSession) {
             LOG.info("Client disconnected. Removing its subscriptions. ClientId={}", clientID);
             // cleanup topic subscriptions
             cleanSession();
+        } else if(disableSession) {
+            LOG.info("Client disconnected. disable session. ClientId={}", clientID);
+            m_sessionsStore.disableSession(this.clientID);
         }
     }
 

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

@@ -148,6 +148,7 @@ public interface ISessionsStore {
      */
     int getSecondPhaseAckPendingMessages(String clientID);
 
+    void disableSession(String clientId);
     void cleanSession(String clientID);
 
     boolean isMultiEndpointSupported();

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

@@ -75,7 +75,7 @@ public class ProtocolProcessor {
             ConnectionDescriptor descriptor = connectionDescriptors.getConnection(session.getClientID());
             try {
                 if (descriptor != null) {
-                    processDisconnect(descriptor.getChannel(), true);
+                    processDisconnect(descriptor.getChannel(), true, false);
                 }
             } catch (InterruptedException e) {
                 e.printStackTrace();
@@ -92,7 +92,7 @@ public class ProtocolProcessor {
                 ConnectionDescriptor descriptor = connectionDescriptors.getConnection(session.getClientID());
                 try {
                     if (descriptor != null) {
-                        processDisconnect(descriptor.getChannel(), true);
+                        processDisconnect(descriptor.getChannel(), true, false);
                     }
                 } catch (InterruptedException e) {
                     e.printStackTrace();
@@ -505,19 +505,19 @@ public class ProtocolProcessor {
        //not use
     }
 
-    public void processDisconnect(Channel channel, boolean clearSession) throws InterruptedException {
+    public void processDisconnect(Channel channel, boolean isDup, boolean isRetain) throws InterruptedException {
         final String clientID = NettyUtils.clientID(channel);
-        LOG.info("Processing DISCONNECT message. CId={}, clearSession={}", clientID, clearSession);
+        LOG.info("Processing DISCONNECT message. CId={}, clearSession={}", clientID, isDup);
         channel.flush();
 
         if (clientID == null) {
-            LOG.error("Error. Cid not exist!!!", clientID, clearSession);
+            LOG.error("Error. Cid not exist!!!", clientID, isDup);
             channel.close();
             return;
         }
 
-        if (!clearSession) {
-            processConnectionLost(clientID, channel, clearSession);
+        if (!isDup && !isRetain) {
+            processConnectionLost(clientID, channel, isDup);
             return;
         }
 
@@ -571,7 +571,7 @@ public class ProtocolProcessor {
 
         //disconnect the session
 
-        m_sessionsStore.sessionForClient(clientID).disconnect(clearSession);
+        m_sessionsStore.sessionForClient(clientID).disconnect(isDup, isRetain);
     }