From 806cd440e36babd1531e45a3304e900e33c9f366 Mon Sep 17 00:00:00 2001 From: heavyrian2012 Date: Wed, 23 Mar 2022 21:45:27 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=AD=A3=E5=9C=A8=E7=BA=BF=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../persistence/MemorySessionStore.java | 4 +-- .../moquette/spi/impl/ProtocolProcessor.java | 25 +++++++++++++------ 2 files changed, 20 insertions(+), 9 deletions(-) diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index da194422..6f8ad020 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -751,7 +751,7 @@ public class MemorySessionStore implements ISessionsStore { public void disableSession(String userId, String clientId) { Session session = sessions.get(clientId); if (session != null && session.getDeleted() == 0 && (userId == null || session.getUsername().equals(userId))) { - mServer.getProcessor().processOffline(session, false); + mServer.getProcessor().processOffline(session, false, null); databaseStore.updateSessionDeleted(session.getUsername(), clientId, 1); ConcurrentSkipListSet sessionSet = getUserSessionSet(session.username); sessionSet.remove(clientId); @@ -773,7 +773,7 @@ public class MemorySessionStore implements ISessionsStore { return; } - mServer.getProcessor().processOffline(session, true); + mServer.getProcessor().processOffline(session, true, null); ConcurrentSkipListSet sessionSet = getUserSessionSet(session.username); sessionSet.remove(clientID); 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 47a647bc..ce0cb230 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -620,19 +620,30 @@ public class ProtocolProcessor { String username = NettyUtils.userName(channel); MemorySessionStore.Session session = m_sessionsStore.getSession(clientID); if(session != null) { - processOffline(session, clearSession); - } - - ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel); - if(connectionDescriptors.removeConnection(oldConnDescr)) { - m_interceptor.notifyClientConnectionLost(clientID, username); + processOffline(session, clearSession, new Runnable() { + @Override + public void run() { + ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel); + if(connectionDescriptors.removeConnection(oldConnDescr)) { + m_interceptor.notifyClientConnectionLost(clientID, username); + } + } + }); + } else { + ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel); + if(connectionDescriptors.removeConnection(oldConnDescr)) { + m_interceptor.notifyClientConnectionLost(clientID, username); + } } } - public void processOffline(MemorySessionStore.Session session, boolean clearSession) { + public void processOffline(MemorySessionStore.Session session, boolean clearSession, Runnable runnable) { if(session != null) { session.refreshLastActiveTime(); forwardOnlineStatusEvent(session.getUsername(), session.getClientID(), session.getPlatform(), clearSession ? UserOnlineStatus.LOGOUT : UserOnlineStatus.OFFLINE, session.getAppName()); + if(runnable != null) { + runnable.run(); + } m_messagesStore.updateUserOnlineSetting(session, false); } }