MQTT 5 client not disconnected on timeout

Author: mario-schwede-hivemq

src/main/java/com/hivemq/extensions/handler/DisconnectInterceptorHandler.java

@@ -16,7 +16,6 @@
 package com.hivemq.extensions.handler;
 
 import com.google.inject.Inject;
-import com.hivemq.bootstrap.ClientConnection;
 import com.hivemq.bootstrap.ClientConnectionContext;
 import com.hivemq.configuration.service.FullConfigurationService;
 import com.hivemq.extension.sdk.api.annotations.NotNull;
@@ -80,12 +79,12 @@ public void handleInboundDisconnect(
             final @NotNull ChannelHandlerContext ctx,
             final @NotNull DISCONNECT disconnect) {
         final Channel channel = ctx.channel();
-        final ClientConnection clientConnection = ClientConnection.of(channel);
-        final String clientId = clientConnection.getClientId();
+        final ClientConnectionContext clientConnectionContext = ClientConnectionContext.of(channel);
+        final String clientId = clientConnectionContext.getClientId();
         if (clientId == null) {
             return;
         }
-        final ClientContextImpl clientContext = clientConnection.getExtensionClientContext();
+        final ClientContextImpl clientContext = clientConnectionContext.getExtensionClientContext();
         if (clientContext == null) {
             ctx.fireChannelRead(disconnect);
             return;
@@ -98,7 +97,7 @@ public void handleInboundDisconnect(
         channel.config().setOption(ChannelOption.ALLOW_HALF_CLOSURE, true);
         final ClientInformation clientInfo = ExtensionInformationUtil.getAndSetClientInformation(channel, clientId);
         final ConnectionInformation connectionInfo = ExtensionInformationUtil.getAndSetConnectionInformation(channel);
-        final Long originalSessionExpiryInterval = clientConnection.getClientSessionExpiryInterval();
+        final Long originalSessionExpiryInterval = clientConnectionContext.getClientSessionExpiryInterval();
         final DisconnectPacketImpl packet = new DisconnectPacketImpl(disconnect);
         final DisconnectInboundInputImpl input = new DisconnectInboundInputImpl(clientInfo, connectionInfo, packet);
         final ExtensionParameterHolder<DisconnectInboundInputImpl> inputHolder = new ExtensionParameterHolder<>(input);

src/main/java/com/hivemq/mqtt/handler/connect/MessageBarrier.java

@@ -23,6 +23,7 @@
 import com.hivemq.mqtt.message.auth.AUTH;
 import com.hivemq.mqtt.message.connack.CONNACK;
 import com.hivemq.mqtt.message.connect.CONNECT;
+import com.hivemq.mqtt.message.disconnect.DISCONNECT;
 import com.hivemq.mqtt.message.reason.Mqtt5ConnAckReasonCode;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelDuplexHandler;
@@ -62,15 +63,13 @@ public void channelRead(final @NotNull ChannelHandlerContext ctx, final @NotNull
         if (msg instanceof Message) {
             if (msg instanceof CONNECT) {
                 connectReceived = true;
-                suspendRead(ctx.channel());
             } else if (!connectReceived) {
                 serverDisconnector.logAndClose(ctx.channel(),
                         "A client (IP: {}) sent other message before CONNECT. Disconnecting client.",
                         "Sent other message before CONNECT");
                 return;
-            } else if (msg instanceof AUTH) {
+            } else if (!(msg instanceof AUTH) && !(msg instanceof DISCONNECT) && !connackSent) {
                 suspendRead(ctx.channel());
-            } else if (!connackSent) {
                 messageQueue.add((Message) msg);
                 return;
             }

src/test/java/com/hivemq/mqtt/handler/connect/MessageBarrierTest.java

@@ -17,6 +17,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.hivemq.bootstrap.ClientConnectionContext;
+import com.hivemq.extension.sdk.api.annotations.NotNull;
 import com.hivemq.logging.EventLog;
 import com.hivemq.mqtt.handler.disconnect.MqttServerDisconnector;
 import com.hivemq.mqtt.handler.disconnect.MqttServerDisconnectorImpl;
@@ -86,9 +87,31 @@ public void test_queue_messages_after_connect() {
         channel.writeInbound(TestMessageUtil.createMqtt3Publish());
         channel.writeInbound(new PINGREQ());
         channel.writeInbound(TestMessageUtil.createMqtt3Publish());
-        channel.writeInbound(new DISCONNECT());
         assertTrue(channel.isActive());
-        assertEquals(6, messageBarrier.getQueue().size());
+        assertEquals(5, messageBarrier.getQueue().size());
+    }
+
+    @Test
+    public void test_disconnect_bypasses_queue_before_connack() {
+        channel.writeInbound(new CONNECT.Mqtt3Builder().withProtocolVersion(ProtocolVersion.MQTTv3_1_1)
+                .withClientIdentifier("clientID")
+                .build());
+
+        final AtomicInteger disconnectCounter = new AtomicInteger(0);
+        channel.pipeline().addAfter(MQTT_MESSAGE_BARRIER, "test", new ChannelDuplexHandler() {
+
+            @Override
+            public void channelRead(final @NotNull ChannelHandlerContext ctx, final @NotNull Object msg) {
+                if (msg instanceof DISCONNECT) {
+                    disconnectCounter.incrementAndGet();
+                }
+            }
+        });
+
+        channel.writeInbound(new DISCONNECT());
+
+        assertEquals(1, disconnectCounter.get());
+        assertEquals(0, messageBarrier.getQueue().size());
     }
 
     @Test