From 739452830d71ad4342b0db77c27ea9768dd909ae Mon Sep 17 00:00:00 2001 From: imhao183 Date: Tue, 28 Apr 2020 07:40:55 +0800 Subject: [PATCH] =?UTF-8?q?=E7=94=A8=E6=88=B7=E9=80=80=E5=87=BA=E6=97=B6?= =?UTF-8?q?=EF=BC=8C=E5=A4=9A=E5=8A=A0=E4=B8=80=E7=A7=8D=E9=80=89=E6=8B=A9?= =?UTF-8?q?=EF=BC=8C=E5=81=9C=E7=94=A8session=E4=B8=8D=E6=B8=85=E9=99=A4se?= =?UTF-8?q?ssion?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../moquette/persistence/MemorySessionStore.java | 12 ++++++++++++ .../moquette/server/netty/NettyMQTTHandler.java | 2 +- .../main/java/io/moquette/spi/ClientSession.java | 5 ++++- .../java/io/moquette/spi/ISessionsStore.java | 1 + .../io/moquette/spi/impl/ProtocolProcessor.java | 16 ++++++++-------- 5 files changed, 26 insertions(+), 10 deletions(-) diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index 191d39a5..71bd00b9 100755 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/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 sessionSet = getUserSessionSet(session.username); + sessionSet.remove(clientId); + sessions.remove(clientId); + } + } + @Override public void cleanSession(String clientID) { LOG.info("Fooooooooo <{}>", clientID); diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index f432dde3..e934f25b 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/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); diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index e523d1fd..653cddf2 100755 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/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); } } diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index 9d190ed4..217ef214 100755 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/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(); 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 d03b439b..de3c9a23 100755 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/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); }