当关闭pc在线通知时可以清理掉历史在线状态

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

@@ -2313,6 +2313,24 @@ public class MemoryMessagesStore implements IMessagesStore {
 
     }
 
+    public void forceCleanOnlineStatus(String userId, String clientId) {
+        if (m_Server.isShutdowning()) {
+            return;
+        }
+
+        if(!mMultiPlatformNotification || m_Server.getStore().sessionsStore().isMultiEndpointSupported()) {
+            WFCMessage.UserSettingEntry pcentry = getUserSetting(userId, kUserSettingPCOnline, "PC");
+            if (pcentry != null && !StringUtil.isNullOrEmpty(pcentry.getValue()) && pcentry.getValue().contains(clientId)) {
+                updateUserSettings(userId, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("PC").setValue("").build(), clientId);
+            }
+
+            WFCMessage.UserSettingEntry padentry = getUserSetting(userId, kUserSettingPCOnline, "Pad");
+            if (padentry != null && !StringUtil.isNullOrEmpty(padentry.getValue()) && padentry.getValue().contains(clientId)) {
+                updateUserSettings(userId, WFCMessage.ModifyUserSettingReq.newBuilder().setScope(kUserSettingPCOnline).setKey("Pad").setValue("").build(), clientId);
+            }
+        }
+    }
+
     @Override
     public void updateUserOnlineSetting(MemorySessionStore.Session session, boolean online) {
         if(!mMultiPlatformNotification) {

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

@@ -729,7 +729,8 @@ public class MemorySessionStore implements ISessionsStore {
         if (session != null) {
             if(!operator.equals(session.getUsername())) {
                 LOG.error("kickoffPCClient failure, user {} don't have client {}", operator, pcClientId);
-                return ErrorCode.ERROR_CODE_NOT_RIGHT;
+                mServer.getProcessor().getMessagesStore().forceCleanOnlineStatus(operator, pcClientId);
+                return ErrorCode.ERROR_CODE_SUCCESS;
             }
             if (session.getPlatform() == ProtoConstants.Platform.Platform_LINUX
                 || session.getPlatform() == ProtoConstants.Platform.Platform_Windows
@@ -746,6 +747,7 @@ public class MemorySessionStore implements ISessionsStore {
             }
         } else {
             LOG.error("Can't find the session for client <{}>", pcClientId);
+            mServer.getProcessor().getMessagesStore().forceCleanOnlineStatus(operator, pcClientId);
         }
         return ErrorCode.ERROR_CODE_SUCCESS;
     }

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

@@ -140,6 +140,8 @@ public interface IMessagesStore {
 
     ErrorCode modifyUserInfo(String userId, WFCMessage.ModifyMyInfoRequest request) throws Exception;
 
+    void forceCleanOnlineStatus(String userId, String clientId);
+
     void updateUserOnlineSetting(MemorySessionStore.Session session, boolean online);
 
     ErrorCode modifyUserStatus(String userId, int status);