From 807debb62250336914f41070c66b0f80ff95a1c3 Mon Sep 17 00:00:00 2001 From: Kevin Berendsen Date: Sun, 7 Aug 2016 13:34:06 +0200 Subject: [PATCH] Terminating the connection if the client failed to receive a CONNACK packet with a non-zero return code. Stated in 3.2.2.3 of MQTT 3.1.1 specificaton. Was checking the wrong channel to see if it was still open or not. The channel that the test was checking on was disconnected a few lines above. --- .../java/io/moquette/spi/impl/ProtocolProcessor.java | 2 ++ .../spi/impl/ProtocolProcessor_CONNECT_Test.java | 11 +++++++++++ 2 files changed, 13 insertions(+) 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 676b3f01..7b3f0128 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -280,6 +280,8 @@ public class ProtocolProcessor { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.BAD_USERNAME_OR_PASSWORD); session.writeAndFlush(okResp); + session.close(); + LOG.info("Client {} failed to connect with bad username or password.", session); } /** diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 27d5b5ee..fddb62e1 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -95,6 +95,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(UNNACEPTABLE_PROTOCOL_VERSION, m_session.readOutbound()); + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } @Test @@ -122,6 +123,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); //TODO verify the call /*verify(mockedMessaging).publish(eq("topic"), eq("Topic message".getBytes()), any(AbstractMessage.QOSType.class), anyBoolean(), eq("123"), any(IoSession.class));*/ @@ -140,6 +142,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); } @Test @@ -154,6 +157,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } @Test @@ -169,6 +173,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } @Test @@ -182,6 +187,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } @Test @@ -197,6 +203,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(BAD_USERNAME_OR_PASSWORD, m_session.readOutbound()); + assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } @Test @@ -210,6 +217,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); } @Test @@ -294,6 +302,7 @@ public class ProtocolProcessor_CONNECT_Test { //Verify assertEqualsConnAck(CONNECTION_ACCEPTED, firstReceiverSession.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", firstReceiverSession.isOpen()); } @@ -306,6 +315,7 @@ public class ProtocolProcessor_CONNECT_Test { connMsg.setCleanSession(false); m_processor.processConnect(m_session, connMsg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); //subscribe SubscribeMessage subscribeMsg = new SubscribeMessage(); @@ -327,6 +337,7 @@ public class ProtocolProcessor_CONNECT_Test { m_session = new EmbeddedChannel(); m_processor.processConnect(m_session, connMsg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); + assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); //verify that the first subscription is still preserved assertTrue(subscriptions.contains(expectedSubscription));