优化多端踢下线的逻辑

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

@@ -407,8 +407,8 @@ public class MemorySessionStore implements ISessionsStore {
 
                         if (remove) {
                             sessions.remove(c);
-                            mServer.getProcessor().kickoffSession(s);
                             it.remove();
+                            mServer.getProcessor().kickoffSession(s);
                         }
                     }
                 }

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

@@ -374,6 +374,25 @@ public class MessagesPublisher {
         }
     }
     
+    public boolean sendOfflineNotify(String clientId) {
+        boolean targetIsActive = this.connectionDescriptors.isConnected(clientId);
+        if (targetIsActive) {
+            ByteBuf payload = Unpooled.buffer();
+            payload.ensureWritable(1).writeBytes("1".getBytes());
+            MqttPublishMessage publishMsg;
+            publishMsg = notRetainedPublish(IMTopic.NotifyOffline, MqttQoS.AT_MOST_ONCE, payload);
+
+            boolean sent = this.messageSender.sendPublish(clientId, publishMsg);
+            if (sent) {
+                return true;
+            }
+        } else {
+            LOG.info("the target {} is not active", clientId);
+        }
+
+        return false;
+    }
+
     private void publishTransparentMessage2Receivers(long messageHead, Collection<String> receivers, int pullType, String exceptClientId) {
         WFCMessage.Message message = m_messagesStore.getMessage(messageHead);
 

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

@@ -54,6 +54,10 @@ class PersistentQueueMessageSender {
 
     boolean sendPublish(ClientSession clientsession, MqttPublishMessage pubMessage) {
         String clientId = clientsession.clientID;
+        return sendPublish(clientId, pubMessage);
+    }
+
+    boolean sendPublish(String clientId, MqttPublishMessage pubMessage) {
         final int messageId = pubMessage.variableHeader().packetId();
         final String topicName = pubMessage.variableHeader().topicName();
         if (LOG.isDebugEnabled()) {

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

@@ -72,15 +72,30 @@ public class ProtocolProcessor {
 
     public void kickoffSession(final MemorySessionStore.Session session) {
         mServer.getImBusinessScheduler().execute(()->{
-            ConnectionDescriptor descriptor = connectionDescriptors.getConnection(session.getClientID());
+            new Thread(() -> {
-            try {
+                messagesPublisher.sendOfflineNotify(session.getClientID());
-                if (descriptor != null) {
+                try {
-                    processDisconnect(descriptor.getChannel(), true, false);
+                    Thread.sleep(3000);
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
                 }
-            } catch (InterruptedException e) {
+
-                e.printStackTrace();
+                mServer.getImBusinessScheduler().execute(()->{
-                Utility.printExecption(LOG, e);
+                    ConnectionDescriptor descriptor = connectionDescriptors.getConnection(session.getClientID());
-            }
+                    try {
+                        if (descriptor != null) {
+                            processDisconnect(descriptor.getChannel(), true, false);
+                        }
+                    } catch (InterruptedException e) {
+                        e.printStackTrace();
+                        Utility.printExecption(LOG, e);
+                    } catch (Exception e) {
+                        e.printStackTrace();
+                        Utility.printExecption(LOG, e);
+                    }
+                    m_messagesStore.updateUserOnlineSetting(session, false);
+                });
+            }).start();
         });
     }
 

broker/src/main/java/win/liyufan/im/IMTopic.java

@@ -16,6 +16,8 @@ public interface IMTopic {
 	String PullMessageTopic = "MP";
 	String NotifyMessageTopic = "MN";
     String NotifyRecallMessageTopic = "RMN";
+    String NotifyOffline = "ROFL";
+
     String BroadcastMessageTopic = "MBC";
 
     String GetUserSettingTopic = "UG";