MQTT 5 client not disconnected on timeout

Author: mario-schwede-hivemq

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