diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java index 4e1a23d4..8f4aab8c 100644 --- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java +++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java @@ -228,21 +228,20 @@ final class MQTTConnection { void handleConnectionLost() { String clientID = NettyUtils.clientID(channel); - if (clientID != null && !clientID.isEmpty()) { - LOG.info("Notifying connection lost event. CId = {}, channel: {}", clientID, channel); - Session session = sessionRegistry.retrieve(clientID); - if (session.hasWill()) { - postOffice.fireWill(session.getWill()); - } - if (session.isClean()) { - sessionRegistry.remove(clientID); - } else { - sessionRegistry.disconnect(clientID); - } - connected = false; + if (clientID == null || clientID.isEmpty()) { + return; } -//?? this breaks test elapseKeepAliveTime -// channel.close().addListener(CLOSE_ON_FAILURE); + LOG.info("Notifying connection lost event. CId: {}, channel: {}", clientID, channel); + Session session = sessionRegistry.retrieve(clientID); + if (session.hasWill()) { + postOffice.fireWill(session.getWill()); + } + if (session.isClean()) { + sessionRegistry.remove(clientID); + } else { + sessionRegistry.disconnect(clientID); + } + connected = false; } void sendConnAck(boolean isSessionAlreadyPresent) { diff --git a/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java b/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java index ffaf1c58..8e543b7e 100644 --- a/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/MoquetteIdleTimeoutHandler.java @@ -37,9 +37,7 @@ public class MoquetteIdleTimeoutHandler extends ChannelDuplexHandler { IdleState e = ((IdleStateEvent) evt).state(); if (e == IdleState.READER_IDLE) { LOG.info("Firing channel inactive event. MqttClientId = {}.", NettyUtils.clientID(ctx.channel())); - // fire a channelInactive to trigger publish of Will -//?? -// ctx.fireChannelInactive(); + // fire a close that then fire channelInactive to trigger publish of Will ctx.close().addListener(CLOSE_ON_FAILURE); } } else { diff --git a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java index b1e9ca2c..bb231030 100644 --- a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -74,7 +74,7 @@ public class ServerLowlevelMessagesIntegrationTests { } @Test - public void elapseKeepAliveTime() throws InterruptedException { + public void elapseKeepAliveTime() { int keepAlive = 2; // secs MqttConnectMessage connectMessage = createConnectMessage("FAKECLNT", keepAlive);