From 23b903e2f282463057862d9288b1d0b6467db5a5 Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 29 Jun 2017 09:23:23 +0200 Subject: [PATCH] - moved and cleaned many part of Paho integration tests to unit tests - fixed bug that on conflicting connect close old channel but not dropped from ConnectionDescriptors cache --- .../impl/PersistentQueueMessageSender.java | 18 +- .../moquette/spi/impl/ProtocolProcessor.java | 8 +- .../subscriptions/SubscriptionsDirectory.java | 4 - .../server/ServerIntegrationPahoTest.java | 589 +++++------------- .../AbstractProtocolProcessorCommonUtils.java | 147 ++++- .../impl/ProtocolProcessor_CONNECT_Test.java | 177 ++---- ...rotocolProcessor_VariousUseCases_Test.java | 377 +++++++++++ 7 files changed, 750 insertions(+), 570 deletions(-) create mode 100644 broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java diff --git a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java index 65963ab8..bb354a06 100644 --- a/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java +++ b/broker/src/main/java/io/moquette/spi/impl/PersistentQueueMessageSender.java @@ -38,23 +38,25 @@ class PersistentQueueMessageSender { String clientId = clientsession.clientID; final int messageId = pubMessage.variableHeader().messageId(); final String topicName = pubMessage.variableHeader().topicName(); + MqttQoS qos = pubMessage.fixedHeader().qosLevel(); if (LOG.isDebugEnabled()) { LOG.debug("Sending PUBLISH message. MessageId={}, CId={}, topic={}, qos={}, payload={}", messageId, - clientId, topicName, DebugUtils.payload2Str(pubMessage.payload())); + clientId, topicName, qos, DebugUtils.payload2Str(pubMessage.payload())); } else { LOG.info("Sending PUBLISH message. MessageId={}, CId={}, topic={}", messageId, clientId, topicName); } boolean messageDelivered = connectionDescriptorStore.sendMessage(pubMessage, messageId, clientId); - MqttQoS qos = pubMessage.fixedHeader().qosLevel(); - if (!messageDelivered && qos != AT_MOST_ONCE && !clientsession.isCleanSession()) { - LOG.warn("PUBLISH message could not be delivered. It will be stored. MessageId={}, CId={}, topic={}, " + if (!messageDelivered) { + if (qos != AT_MOST_ONCE && !clientsession.isCleanSession()) { + LOG.warn("PUBLISH message could not be delivered. It will be stored. MessageId={}, CId={}, topic={}, " + "qos={}, cleanSession={}", messageId, clientId, topicName, qos, false); - clientsession.enqueue(asStoredMessage(pubMessage)); - } else { - LOG.warn("PUBLISH message could not be delivered. It will be discarded. MessageId={}, CId={}, topic={}, " + - "qos={}, cleanSession={}", messageId, clientId, topicName, qos, true); + clientsession.enqueue(asStoredMessage(pubMessage)); + } else { + LOG.warn("PUBLISH message could not be delivered. It will be discarded. MessageId={}, CId={}, topic={}, " + + "qos={}, cleanSession={}", messageId, clientId, topicName, qos, true); + } } } } 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 6acb919c..c404566b 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -269,9 +269,11 @@ public class ProtocolProcessor { msg.variableHeader().isCleanSession()); ConnectionDescriptor existing = this.connectionDescriptors.addConnection(descriptor); if (existing != null) { - LOG.info("The client ID is being used in an existing connection. It will be closed. CId={}", clientId); + LOG.info("Client ID is being used in an existing connection, force to be closed. CId={}", clientId); existing.abort(); - return; + //return; + this.connectionDescriptors.removeConnection(existing); + this.connectionDescriptors.addConnection(descriptor); } initializeKeepAliveTimeout(channel, msg, clientId); @@ -298,7 +300,7 @@ public class ProtocolProcessor { channel.close(); } - LOG.info("The CONNECT message has been processed. CId={}, username={}", clientId, payload.userName()); + LOG.info("CONNECT message processed CId={}, username={}", clientId, payload.userName()); } private MqttConnAckMessage connAck(MqttConnectReturnCode returnCode) { diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java index 33228f92..c2e5cdd1 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectory.java @@ -218,10 +218,6 @@ public class SubscriptionsDirectory { return new ArrayList<>(subsForClient.values()); } - public boolean contains(Subscription sub) { - return !matches(sub.topicFilter).isEmpty(); - } - public int size() { return subscriptions.get().size(); } diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index 8cd39f07..73a2f70c 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java @@ -83,94 +83,6 @@ public class ServerIntegrationPahoTest { m_server.stopServer(); } - @Test - public void connectWithCredentials() throws Exception { - LOG.info("*** connectWithCredentials ***"); - m_client = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); - MqttConnectOptions options = new MqttConnectOptions(); - options.setUserName("testuser"); - options.setPassword("passwd".toCharArray()); - m_client.connect(options); - - assertTrue(true); - } - - @Test - public void testSubscribe() throws Exception { - LOG.info("*** testSubscribe ***"); - m_client.connect(); - m_client.subscribe("/topic", 0); - - MqttMessage message = new MqttMessage("Hello world!!".getBytes()); - message.setQos(0); - message.setRetained(false); - m_client.publish("/topic", message); - - assertEquals("/topic", m_messagesCollector.getTopic()); - } - - @Test - public void testCleanSession_maintainClientSubscriptions() throws Exception { - LOG.info("*** testCleanSession_maintainClientSubscriptions ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 0); - m_client.disconnect(); - - // reconnect and publish - MqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); - m_messagesCollector = new MessageCollector(); - anotherClient.setCallback(m_messagesCollector); - anotherClient.connect(options); - anotherClient.publish("/topic", "Test my payload".getBytes(), 0, false); - - assertEquals("/topic", m_messagesCollector.getTopic()); - } - - @Test - public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() throws Exception { - LOG.info("*** testCleanSession_maintainClientSubscriptions_againstClientDestruction ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 0); - m_client.disconnect(); - - // reconnect and publish - m_client.connect(options); - m_client.publish("/topic", "Test my payload".getBytes(), 0, false); - - assertEquals("/topic", m_messagesCollector.getTopic()); - } - - /** - * Check that after a client has connected with clean session false, subscribed to some topic - * and exited, if it reconnects with clean session true, the m_server correctly cleanup every - * previous subscription - */ - @Test - public void testCleanSession_correctlyClientSubscriptions() throws Exception { - LOG.info("*** testCleanSession_correctlyClientSubscriptions ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 0); - m_client.disconnect(); - - // reconnect and publish - // options.setCleanSession(false); - m_client = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); - MessageCollector subListener = new MessageCollector(); - m_client.setCallback(subListener); - m_client.connect(options); - m_client.publish("/topic", "Test my payload".getBytes(), 0, false); - // Ugly, we need to get self notified after the publish - assertEquals("Test my payload", new String(subListener.waitMessage(1).getPayload())); - - // close - m_client.disconnect(); - } @Ignore("This test hasn't any meaning using in memory storage service") @Test @@ -193,187 +105,6 @@ public class ServerIntegrationPahoTest { assertEquals("/topic", m_messagesCollector.getTopic()); } - @Test - public void testRetain_maintainMessage_againstClientDestruction() throws Exception { - LOG.info("*** testRetain_maintainMessage_againstClientDestruction ***"); - m_client.connect(); - m_client.publish("/topic", "Test my payload".getBytes(), 1, true); - m_client.disconnect(); - - // reconnect and publish - m_client.connect(); - m_client.subscribe("/topic", 0); - - assertEquals("/topic", m_messagesCollector.getTopic()); - } - - @Test - public void testUnsubscribe_do_not_notify_anymore_same_session() throws Exception { - LOG.info("*** testUnsubscribe_do_not_notify_anymore_same_session ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 0); - m_client.publish("/topic", "Test my payload".getBytes(), 0, false); - // m_client.disconnect(); - assertEquals("/topic", m_messagesCollector.getTopic()); - - m_client.unsubscribe("/topic"); - m_messagesCollector.reinit(); - m_client.publish("/topic", "Test my payload".getBytes(), 0, false); - - assertNull(m_messagesCollector.getMessageImmediate()); - } - - @Test - public void testUnsubscribe_do_not_notify_anymore_new_session() throws Exception { - LOG.info("*** testUnsubscribe_do_not_notify_anymore_new_session ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 0); - m_client.publish("/topic", "Test my payload".getBytes(), 0, false); - // m_client.disconnect(); - assertEquals("/topic", m_messagesCollector.getTopic()); - - m_client.unsubscribe("/topic"); - m_client.disconnect(); - - m_messagesCollector.reinit(); - m_client.connect(options); - m_client.publish("/topic", "Test my payload".getBytes(), 0, false); - - assertNull(m_messagesCollector.getMessageImmediate()); - } - - @Test - public void testPublishWithQoS1() throws Exception { - LOG.info("*** testPublishWithQoS1 ***"); - m_client.connect(); - m_client.subscribe("/topic", 1); - m_client.publish("/topic", "Hello MQTT".getBytes(), 1, false); - m_client.disconnect(); - - // reconnect and publish - MqttMessage message = m_messagesCollector.waitMessage(1); - assertEquals("Hello MQTT", message.toString()); - assertEquals(1, message.getQos()); - } - - @Test - public void testPublishWithQoS1_notCleanSession() throws Exception { - LOG.info("*** testPublishWithQoS1_notCleanSession ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 1); - m_client.disconnect(); - - // publish a QoS 1 message another client publish a message on the topic - publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 1); - - m_client.connect(options); - - assertEquals("Hello MQTT", m_messagesCollector.waitMessage(1).toString()); - } - - @Test - public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws Exception { - LOG.info("*** checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 1); - m_client.disconnect(); - - m_client.connect(options); - m_client.subscribe("/topic", 1); - - // publish a QoS 1 message another client publish a message on the topic - publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 1); - - // Verify that after a reconnection the client receive the message - MqttMessage message = m_messagesCollector.waitMessage(1); - assertNotNull(message); - assertEquals("Hello MQTT", message.toString()); - } - - private void publishFromAnotherClient(String topic, byte[] payload, int qos) throws Exception { - IMqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClientPUB", s_pubDataStore); - anotherClient.connect(); - anotherClient.publish(topic, payload, qos, false); - anotherClient.disconnect(); - } - - @Test - public void testPublishWithQoS2() throws Exception { - LOG.info("*** testPublishWithQoS2 ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 2); - m_client.disconnect(); - - // publish a QoS 1 message another client publish a message on the topic - publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 2); - m_messagesCollector.reinit(); - m_client.connect(options); - - MqttMessage message = m_messagesCollector.waitMessage(1); - assertEquals("Hello MQTT", message.toString()); - assertEquals(2, message.getQos()); - } - - @Test - public void testPublishReceiveWithQoS2() throws Exception { - LOG.info("*** testPublishReceiveWithQoS2 ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 2); - m_client.disconnect(); - - // publish a QoS 2 message another client publish a message on the topic - publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 2); - m_messagesCollector.reinit(); - m_client.connect(options); - - assertNotNull(m_messagesCollector); - MqttMessage message = m_messagesCollector.waitMessage(1); - assertNotNull(message); - assertEquals("Hello MQTT", message.toString()); - } - - @Test - public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() throws Exception { - LOG.info("*** avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1, issue #16 ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 1); - m_client.disconnect(); - - publishFromAnotherClient("/topic", "Hello MQTT 1".getBytes(), 1); - m_messagesCollector.reinit(); - m_client.connect(options); - - assertNotNull(m_messagesCollector); - MqttMessage message = m_messagesCollector.waitMessage(1); - assertNotNull(message); - assertEquals("Hello MQTT 1", message.toString()); - m_client.disconnect(); - - // publish other message - publishFromAnotherClient("/topic", "Hello MQTT 2".getBytes(), 1); - - // reconnect the second time - m_messagesCollector.reinit(); - m_client.connect(options); - assertNotNull(m_messagesCollector); - message = m_messagesCollector.waitMessage(1); - assertNotNull(message); - assertEquals("Hello MQTT 2", message.toString()); - } /** * subscriber A connect and subscribe on "a/b" QoS 1 subscriber B connect and subscribe on "a/+" @@ -460,170 +191,170 @@ public class ServerIntegrationPahoTest { assertEquals("Hello", new String(messageOnB.getPayload())); } - @Test - public void testForceClientDisconnection_issue116() throws Exception { - LOG.info("*** testForceClientDisconnection_issue118 ***"); - MessageCollector cbSubscriber1 = new MessageCollector(); - MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1); - LOG.info("Connected 'subscriber' first time"); - clientXA.subscribe("topic", 0); - LOG.info("Subscribed 'topic' from 'subscriber' first time"); +// @Test +// public void testForceClientDisconnection_issue116() throws Exception { +// LOG.info("*** testForceClientDisconnection_issue118 ***"); +// MessageCollector cbSubscriber1 = new MessageCollector(); +// MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1); +// LOG.info("Connected 'subscriber' first time"); +// clientXA.subscribe("topic", 0); +// LOG.info("Subscribed 'topic' from 'subscriber' first time"); +// +// MqttClient clientXB = createClient("publisher", "X"); +// LOG.info("Connected 'publisher' first time"); +// clientXB.publish("topic", "Hello".getBytes(), 2, false); +// LOG.info("Published on 'topic' from 'publisher' first time"); +// +// LOG.info("Creating second new 'subscriber'"); +// MessageCollector cbSubscriber2 = new MessageCollector(); +// MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2); +// LOG.info("Connected 'subscriber' second time"); +// clientYA.subscribe("topic", 0); +// +// MqttClient clientYB = createClient("publisher", "Y"); +// clientYB.publish("topic", "Hello 2".getBytes(), 2, true); +// +// // Verify that the second subscriber client get notified and not the first. +// assertTrue(cbSubscriber1.connectionLost()); +// assertEquals("Hello 2", new String(cbSubscriber2.waitMessage(1).getPayload())); +// } - MqttClient clientXB = createClient("publisher", "X"); - LOG.info("Connected 'publisher' first time"); - clientXB.publish("topic", "Hello".getBytes(), 2, false); - LOG.info("Published on 'topic' from 'publisher' first time"); +// @Test +// public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws Exception { +// LOG.info("*** testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142 ***"); +// MqttConnectOptions options = new MqttConnectOptions(); +// options.setCleanSession(false); +// m_client.connect(options); +// m_client.subscribe("/topic", 0); +// m_client.publish("/topic", "Hello".getBytes(), 0, true); +// m_client.disconnect(); +// +// // second loop +// m_client.connect(options); +// m_client.subscribe("/topic", 0); +// m_client.publish("/topic", "Hello".getBytes(), 0, true); +// m_client.disconnect(); // this should give timeout +// +// assertFalse("after a disconnect the client should be disconnected", m_client.isConnected()); +// } - LOG.info("Creating second new 'subscriber'"); - MessageCollector cbSubscriber2 = new MessageCollector(); - MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2); - LOG.info("Connected 'subscriber' second time"); - clientYA.subscribe("topic", 0); +// @Test +// public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws Exception { +// LOG.info("*** checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1 ***"); +// // MessageCollector m_messagesCollector = new MessageCollector(); +// // m_client.setCallback(m_messagesCollector); +// m_publisher.connect(); +// +// MqttConnectOptions options = new MqttConnectOptions(); +// options.setCleanSession(false); +// m_client.connect(options); +// m_client.subscribe("/topic", 1); +// +// // force the publisher to send +// m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false); +// +// // read the first message and drop the connection +// MqttMessage msg = m_messagesCollector.waitMessage(1); +// assertEquals("Hello world MQTT!!-1", new String(msg.getPayload())); +// m_client.disconnect(); +// +// m_publisher.publish("/topic", "Hello world MQTT!!-2".getBytes(), 1, false); +// m_publisher.publish("/topic", "Hello world MQTT!!-3".getBytes(), 1, false); +// +// // reconnect and expect to receive the hello 2 message +// m_client.connect(options); +// msg = m_messagesCollector.waitMessage(1); +// assertEquals("Hello world MQTT!!-2", new String(msg.getPayload())); +// +// msg = m_messagesCollector.waitMessage(1); +// assertEquals("Hello world MQTT!!-3", new String(msg.getPayload())); +// m_client.disconnect(); +// } - MqttClient clientYB = createClient("publisher", "Y"); - clientYB.publish("topic", "Hello 2".getBytes(), 2, true); +// @Test +// public void checkReplayStoredPublish_forNoCleanSession_qos1() throws Exception { +// LOG.info("*** checkReplayStoredPublish_forNoCleanSession_qos1 ***"); +// MessageCollector messagesCollector = new MessageCollector(); +// m_client.setCallback(messagesCollector); +// m_publisher.connect(); +// +// MqttConnectOptions options = new MqttConnectOptions(); +// options.setCleanSession(false); +// m_client.connect(options); +// m_client.subscribe("/topic", 1); +// m_client.disconnect(); +// +// // force the publisher to send +// m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false); +// +// // reconnect and expect to receive the hello 2 message +// m_client.connect(options); +// m_client.subscribe("/topic", 1); +// MqttMessage msg = messagesCollector.waitMessage(1); +// assertNotNull(msg); +// assertEquals("Hello world MQTT!!-1", new String(msg.getPayload())); +// } - // Verify that the second subscriber client get notified and not the first. - assertTrue(cbSubscriber1.connectionLost()); - assertEquals("Hello 2", new String(cbSubscriber2.waitMessage(1).getPayload())); - } - - @Test - public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws Exception { - LOG.info("*** testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142 ***"); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 0); - m_client.publish("/topic", "Hello".getBytes(), 0, true); - m_client.disconnect(); - - // second loop - m_client.connect(options); - m_client.subscribe("/topic", 0); - m_client.publish("/topic", "Hello".getBytes(), 0, true); - m_client.disconnect(); // this should give timeout - - assertFalse("after a disconnect the client should be disconnected", m_client.isConnected()); - } - - @Test - public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws Exception { - LOG.info("*** checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1 ***"); - // MessageCollector m_messagesCollector = new MessageCollector(); - // m_client.setCallback(m_messagesCollector); - m_publisher.connect(); - - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 1); - - // force the publisher to send - m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false); - - // read the first message and drop the connection - MqttMessage msg = m_messagesCollector.waitMessage(1); - assertEquals("Hello world MQTT!!-1", new String(msg.getPayload())); - m_client.disconnect(); - - m_publisher.publish("/topic", "Hello world MQTT!!-2".getBytes(), 1, false); - m_publisher.publish("/topic", "Hello world MQTT!!-3".getBytes(), 1, false); - - // reconnect and expect to receive the hello 2 message - m_client.connect(options); - msg = m_messagesCollector.waitMessage(1); - assertEquals("Hello world MQTT!!-2", new String(msg.getPayload())); - - msg = m_messagesCollector.waitMessage(1); - assertEquals("Hello world MQTT!!-3", new String(msg.getPayload())); - m_client.disconnect(); - } - - @Test - public void checkReplayStoredPublish_forNoCleanSession_qos1() throws Exception { - LOG.info("*** checkReplayStoredPublish_forNoCleanSession_qos1 ***"); - MessageCollector messagesCollector = new MessageCollector(); - m_client.setCallback(messagesCollector); - m_publisher.connect(); - - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("/topic", 1); - m_client.disconnect(); - - // force the publisher to send - m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false); - - // reconnect and expect to receive the hello 2 message - m_client.connect(options); - m_client.subscribe("/topic", 1); - MqttMessage msg = messagesCollector.waitMessage(1); - assertNotNull(msg); - assertEquals("Hello world MQTT!!-1", new String(msg.getPayload())); - } - - /** - * subscriber connect and subscribe on "topic" subscriber disconnects publisher connects and - * send two message "hello1" "hello2" to "topic" subscriber connects again and receive "hello1" - * "hello2" - */ - @Test - public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() throws Exception { - LOG.info("*** checkQoS2SubscriberDisconnectReceivePersistedPublishes ***"); - MessageCollector messagesCollector = new MessageCollector(); - m_client.setCallback(messagesCollector); - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("topic", 2); - m_client.disconnect(); - - m_publisher.connect(); - m_publisher.publish("topic", "Hello1".getBytes(), 2, true); - m_publisher.publish("topic", "Hello2".getBytes(), 2, true); - m_publisher.disconnect(); - - // subscriber reconnects - m_client.connect(options); - // subscriber should receive the 2 messages missed - MqttMessage msg = messagesCollector.waitMessage(1); - assertNotNull(msg); - assertEquals("Hello1", new String(msg.getPayload())); - msg = messagesCollector.waitMessage(1); - assertNotNull(msg); - assertEquals("Hello2", new String(msg.getPayload())); - } - - /** - * subscriber connect and subscribe on "a/b" QoS 1 and "a/+" QoS 2 publisher connects and send a - * message "hello" on "a/b" subscriber must receive only a single message not twice - */ - @Test - public void checkSinglePublishOnOverlappingSubscriptions() throws Exception { - LOG.info("*** checkSinglePublishOnOverlappingSubscriptions ***"); - MessageCollector messagesCollector = new MessageCollector(); - m_client.setCallback(messagesCollector); - m_publisher.connect(); - - MqttConnectOptions options = new MqttConnectOptions(); - options.setCleanSession(false); - m_client.connect(options); - m_client.subscribe("a/+", 2); - m_client.subscribe("a/b", 1); - - // force the publisher to send - m_publisher.publish("a/b", "Hello world MQTT!!".getBytes(), 1, false); - - // reconnect and expect to receive the hello 2 message - MqttMessage msg = messagesCollector.waitMessage(1); - assertEquals("Hello world MQTT!!", new String(msg.getPayload())); - - // try to listen a second publish - msg = messagesCollector.waitMessage(1); - assertNull(msg); - } +// /** +// * subscriber connect and subscribe on "topic" subscriber disconnects publisher connects and +// * send two message "hello1" "hello2" to "topic" subscriber connects again and receive "hello1" +// * "hello2" +// */ +// @Test +// public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() throws Exception { +// LOG.info("*** checkQoS2SubscriberDisconnectReceivePersistedPublishes ***"); +// MessageCollector messagesCollector = new MessageCollector(); +// m_client.setCallback(messagesCollector); +// MqttConnectOptions options = new MqttConnectOptions(); +// options.setCleanSession(false); +// m_client.connect(options); +// m_client.subscribe("topic", 2); +// m_client.disconnect(); +// +// m_publisher.connect(); +// m_publisher.publish("topic", "Hello1".getBytes(), 2, true); +// m_publisher.publish("topic", "Hello2".getBytes(), 2, true); +// m_publisher.disconnect(); +// +// // subscriber reconnects +// m_client.connect(options); +// // subscriber should receive the 2 messages missed +// MqttMessage msg = messagesCollector.waitMessage(1); +// assertNotNull(msg); +// assertEquals("Hello1", new String(msg.getPayload())); +// msg = messagesCollector.waitMessage(1); +// assertNotNull(msg); +// assertEquals("Hello2", new String(msg.getPayload())); +// } +// +// /** +// * subscriber connect and subscribe on "a/b" QoS 1 and "a/+" QoS 2 publisher connects and send a +// * message "hello" on "a/b" subscriber must receive only a single message not twice +// */ +// @Test +// public void checkSinglePublishOnOverlappingSubscriptions() throws Exception { +// LOG.info("*** checkSinglePublishOnOverlappingSubscriptions ***"); +// MessageCollector messagesCollector = new MessageCollector(); +// m_client.setCallback(messagesCollector); +// m_publisher.connect(); +// +// MqttConnectOptions options = new MqttConnectOptions(); +// options.setCleanSession(false); +// m_client.connect(options); +// m_client.subscribe("a/+", 2); +// m_client.subscribe("a/b", 1); +// +// // force the publisher to send +// m_publisher.publish("a/b", "Hello world MQTT!!".getBytes(), 1, false); +// +// // reconnect and expect to receive the hello 2 message +// MqttMessage msg = messagesCollector.waitMessage(1); +// assertEquals("Hello world MQTT!!", new String(msg.getPayload())); +// +// // try to listen a second publish +// msg = messagesCollector.waitMessage(1); +// assertNull(msg); +// } protected MqttClient createClient(String clientName, String storeSuffix) throws MqttException { return createClient(clientName, storeSuffix, null); diff --git a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java index 3903526b..44f51698 100644 --- a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java +++ b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java @@ -20,21 +20,24 @@ import io.moquette.persistence.MemoryStorageService; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; +import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory; import io.moquette.spi.impl.subscriptions.Topic; import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.mqtt.*; import java.util.*; import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED; +import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; +import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE; import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; abstract class AbstractProtocolProcessorCommonUtils { @@ -82,20 +85,32 @@ abstract class AbstractProtocolProcessorCommonUtils { } protected void verifyNoPublishIsReceived() { - assertNull("No out messages from the processor", m_channel.readOutbound()); + assertNull("Received an out message from processor while not expected", m_channel.readOutbound()); } protected void subscribe(String topic, MqttQoS desiredQos) { + subscribe(this.m_channel, topic, desiredQos); + } + + protected void subscribe(EmbeddedChannel channel, String topic, MqttQoS desiredQos) { MqttSubscribeMessage subscribe = MqttMessageBuilders.subscribe() .addSubscription(desiredQos, topic) .messageId(1) .build(); - this.m_processor.processSubscribe(m_channel, subscribe); - MqttSubAckMessage subAck = m_channel.readOutbound(); + this.m_processor.processSubscribe(channel, subscribe); + MqttSubAckMessage subAck = channel.readOutbound(); assertEquals(desiredQos.value(), (int) subAck.payload().grantedQoSLevels().get(0)); - Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, new Topic(topic), desiredQos); - assertTrue(subscriptions.contains(expectedSubscription)); + final String clientId = NettyUtils.clientID(channel); + Subscription expectedSubscription = new Subscription(clientId, new Topic(topic), desiredQos); + verifySubscriptionExists(channel, m_sessionStore, expectedSubscription); + } + + protected static void verifySubscriptionExists(Channel channel, ISessionsStore sessionsStore, Subscription expectedSubscription) { + final String clientId = NettyUtils.clientID(channel); + final Subscription subscription = sessionsStore.subscriptionStore() + .getSubscription(new ClientTopicCouple(clientId, expectedSubscription.getTopicFilter())); + assertEquals(expectedSubscription, subscription); } protected MqttSubAckMessage subscribeWithoutVerify(String topic, MqttQoS desiredQos) { @@ -116,14 +131,28 @@ abstract class AbstractProtocolProcessorCommonUtils { } protected void unsubscribe(String topic) { + final int messageId = 1; MqttUnsubscribeMessage msg = MqttMessageBuilders.unsubscribe() .addTopicFilter(topic) - .messageId(1) + .messageId(messageId) .build(); m_processor.processUnsubscribe(m_channel, msg); } + protected void unsubscribeAndVerifyAck(String topic) { + final int messageId = 1; + MqttUnsubscribeMessage msg = MqttMessageBuilders.unsubscribe() + .addTopicFilter(topic) + .messageId(messageId) + .build(); + + m_processor.processUnsubscribe(m_channel, msg); + + MqttUnsubAckMessage unsubAckMessageAck = m_channel.readOutbound(); + assertEquals("Unsubscribe must be accepted", messageId, unsubAckMessageAck.variableHeader().messageId()); + } + protected void internalPublishNotRetainedTo(String topic) { internalPublishTo(topic, AT_MOST_ONCE, false); } @@ -142,7 +171,11 @@ abstract class AbstractProtocolProcessorCommonUtils { } protected void publishToAs(String clientId, String topic, MqttQoS qos, boolean retained) { - NettyUtils.userName(m_channel, clientId); + publishToAs(m_channel, clientId, topic, qos, retained); + } + + protected void publishToAs(EmbeddedChannel channel, String clientId, String topic, MqttQoS qos, boolean retained) { + NettyUtils.userName(channel, clientId); MqttPublishMessage publish = MqttMessageBuilders.publish() .topicName(topic) .retained(retained) @@ -152,14 +185,55 @@ abstract class AbstractProtocolProcessorCommonUtils { } protected void publishToAs(String clientId, String topic, MqttQoS qos, int messageId, boolean retained) { - NettyUtils.userName(m_channel, clientId); + publishToAs(this.m_channel, clientId, topic, qos, messageId, retained); + } + + protected void publishToAs(EmbeddedChannel channel, String clientId, String topic, MqttQoS qos, int messageId, boolean retained) { + publishToAs(channel, clientId, topic, HELLO_WORLD_MQTT, qos, messageId, retained); + } + + protected void publishToAs(EmbeddedChannel channel, String clientId, String topic, String payload, MqttQoS qos, int messageId, boolean retained) { + NettyUtils.userName(channel, clientId); MqttPublishMessage publish = MqttMessageBuilders.publish() .topicName(topic) .retained(retained) .messageId(messageId) .qos(qos) - .payload(Unpooled.copiedBuffer(HELLO_WORLD_MQTT.getBytes())).build(); - this.m_processor.processPublish(m_channel, publish); + .payload(Unpooled.copiedBuffer(payload.getBytes())).build(); + this.m_processor.processPublish(channel, publish); + } + + protected void publishQoS2ToAs(EmbeddedChannel channel, String clientId, String topic, int messageId, boolean retained) { + publishQoS2ToAs(channel, clientId, topic, HELLO_WORLD_MQTT, messageId, retained); + } + + protected void publishQoS2ToAs(EmbeddedChannel channel, String clientId, String topic, String payload, int messageId, boolean retained) { + NettyUtils.userName(channel, clientId); + MqttPublishMessage publish = MqttMessageBuilders.publish() + .topicName(topic) + .retained(retained) + .messageId(messageId) + .qos(MqttQoS.EXACTLY_ONCE) + .payload(Unpooled.copiedBuffer(payload.getBytes())).build(); + this.m_processor.processPublish(channel, publish); + + verifyPubrecIsReceived(channel, messageId); + + MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0); + MqttMessage pubRel = new MqttMessage(mqttFixedHeader, from(messageId)); + this.m_processor.processPubRel(channel, pubRel); + + verifyPubCompIsReceived(channel, messageId); + } + + private void verifyPubrecIsReceived(EmbeddedChannel channel, int messageId) { + final MqttMessage pubRec = channel.readOutbound(); + assertEquals(messageId, Utils.messageId(pubRec)); + } + + private void verifyPubCompIsReceived(EmbeddedChannel channel, int messageId) { + final MqttMessage pubComp = channel.readOutbound(); + assertEquals(messageId, Utils.messageId(pubComp)); } @@ -168,11 +242,15 @@ abstract class AbstractProtocolProcessorCommonUtils { } protected void connectAsClient(String clientId) { + connectAsClient(m_channel, clientId); + } + + protected void connectAsClient(EmbeddedChannel channel, String clientId) { MqttConnectMessage connectMessage = MqttMessageBuilders.connect() .clientId(clientId) .build(); - this.m_processor.processConnect(m_channel, connectMessage); - MqttConnAckMessage connAck = m_channel.readOutbound(); + this.m_processor.processConnect(channel, connectMessage); + MqttConnAckMessage connAck = channel.readOutbound(); assertEquals("Connect must be accepted", CONNECTION_ACCEPTED, connAck.variableHeader().connectReturnCode()); } @@ -192,8 +270,12 @@ abstract class AbstractProtocolProcessorCommonUtils { } protected void connectNoCleanSession() { + connectNoCleanSession(FAKE_CLIENT_ID); + } + + protected void connectNoCleanSession(String clientId) { MqttConnectMessage connectMessage = MqttMessageBuilders.connect() - .clientId(FAKE_CLIENT_ID) + .clientId(clientId) .cleanSession(false) .build(); this.m_processor.processConnect(m_channel, connectMessage); @@ -201,13 +283,46 @@ abstract class AbstractProtocolProcessorCommonUtils { assertEquals("Connect must be accepted", CONNECTION_ACCEPTED, connAck.variableHeader().connectReturnCode()); } + protected void connectWithCleanSession(String clientId) { + MqttConnectMessage connectMessage = MqttMessageBuilders.connect() + .clientId(clientId) + .cleanSession(true) + .build(); + this.m_processor.processConnect(m_channel, connectMessage); + MqttConnAckMessage connAck = m_channel.readOutbound(); + assertEquals("Connect must be accepted", CONNECTION_ACCEPTED, connAck.variableHeader().connectReturnCode()); + } + protected void disconnect() throws InterruptedException { - this.m_processor.processDisconnect(this.m_channel); + disconnect(this.m_channel); + } + + protected void disconnect(EmbeddedChannel channel) throws InterruptedException { + this.m_processor.processDisconnect(channel); } protected void verifyPublishIsReceived() { - final MqttPublishMessage publishReceived = m_channel.readOutbound(); + verifyPublishIsReceived(m_channel); + } + + protected void verifyPublishIsReceived(EmbeddedChannel channel) { + final MqttPublishMessage publishReceived = channel.readOutbound(); String payloadMessage = new String(publishReceived.payload().array()); assertEquals("Sent and received payload must be identical", HELLO_WORLD_MQTT, payloadMessage); } + + protected void verifyPublishIsReceived(MqttQoS expectedQoS) { + verifyPublishIsReceived(HELLO_WORLD_MQTT, expectedQoS); + } + + protected void verifyPublishIsReceived(String expectedPayload, MqttQoS expectedQoS) { + verifyPublishIsReceived(this.m_channel, expectedPayload, expectedQoS); + } + + protected void verifyPublishIsReceived(EmbeddedChannel channel, String expectedPayload, MqttQoS expectedQoS) { + final MqttPublishMessage publishReceived = channel.readOutbound(); + String payloadMessage = new String(publishReceived.payload().array()); + assertEquals("Sent and received payload must be identical", expectedPayload, payloadMessage); + assertEquals("Expected QoS don't match", expectedQoS, publishReceived.fixedHeader().qosLevel()); + } } 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 4c94d13c..d51d9eed 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 @@ -30,14 +30,12 @@ import io.netty.handler.codec.mqtt.*; import org.junit.Before; import org.junit.Test; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsConnAck; import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsSubAck; +import static io.moquette.spi.impl.ProtocolProcessorTest.*; import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*; +import static java.util.Collections.singleton; +import static java.util.Collections.singletonMap; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -63,23 +61,13 @@ public class ProtocolProcessor_CONNECT_Test { m_sessionStore = memStorage.sessionsStore(); // m_messagesStore.initStore(); - Set clientIds = new HashSet<>(); - clientIds.add(ProtocolProcessorTest.FAKE_CLIENT_ID); - Map users = new HashMap<>(); - users.put(ProtocolProcessorTest.TEST_USER, ProtocolProcessorTest.TEST_PWD); - m_mockAuthenticator = new MockAuthenticator(clientIds, users); + m_mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD)); subscriptions = new SubscriptionsDirectory(); subscriptions.init(m_sessionStore); m_processor = new ProtocolProcessor(); - m_processor.init( - subscriptions, - m_messagesStore, - m_sessionStore, - m_mockAuthenticator, - true, - new PermitAllAuthorizator(), - ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); } @Test @@ -95,7 +83,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void testWill() { - MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).willFlag(true) + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).willFlag(true) .willTopic("topic").willMessage("Topic message").build(); // Exercise @@ -114,8 +102,8 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void validAuthentication() { - MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID) - .username(ProtocolProcessorTest.TEST_USER).password(ProtocolProcessorTest.TEST_PWD).build(); + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .username(TEST_USER).password(TEST_PWD).build(); // Exercise m_processor.processConnect(m_session, msg); @@ -127,8 +115,9 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void noPasswdAuthentication() { - MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID) - .username(ProtocolProcessorTest.TEST_USER).build(); + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .username(TEST_USER) + .build(); // Exercise m_processor.processConnect(m_session, msg); @@ -140,8 +129,10 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void invalidAuthentication() { - MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID) - .username(ProtocolProcessorTest.TEST_USER + "_fake").password(ProtocolProcessorTest.TEST_PWD).build(); + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .username(TEST_USER + "_fake") + .password(TEST_PWD) + .build(); // Exercise m_processor.processConnect(m_session, msg); @@ -153,15 +144,8 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void prohibitAnonymousClient() { - MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).build(); - m_processor.init( - subscriptions, - m_messagesStore, - m_sessionStore, - m_mockAuthenticator, - false, - new PermitAllAuthorizator(), - ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build(); + reinitProcessorProhibitingAnonymousClients(); // Exercise m_processor.processConnect(m_session, msg); @@ -171,18 +155,17 @@ public class ProtocolProcessor_CONNECT_Test { assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } + protected void reinitProcessorProhibitingAnonymousClients() { + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + } + @Test public void prohibitAnonymousClient_providingUsername() { - MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID) - .username(ProtocolProcessorTest.TEST_USER + "_fake").build(); - m_processor.init( - subscriptions, - m_messagesStore, - m_sessionStore, - m_mockAuthenticator, - false, - new PermitAllAuthorizator(), - ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .username(TEST_USER + "_fake") + .build(); + reinitProcessorProhibitingAnonymousClients(); // Exercise m_processor.processConnect(m_session, msg); @@ -194,15 +177,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void acceptAnonymousClient() { - MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).build(); - m_processor.init( - subscriptions, - m_messagesStore, - m_sessionStore, - m_mockAuthenticator, - true, - new PermitAllAuthorizator(), - ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build(); // Exercise m_processor.processConnect(m_session, msg); @@ -214,17 +189,8 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void connectWithCleanSessionUpdateClientSession() throws InterruptedException { - m_processor.init( - subscriptions, - m_messagesStore, - m_sessionStore, - m_mockAuthenticator, - true, - new PermitAllAuthorizator(), - ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); - // first connect with clean session true - MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).cleanSession(true).build(); + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).cleanSession(true).build(); m_processor.processConnect(m_session, msg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); m_processor.processDisconnect(m_session); @@ -232,14 +198,14 @@ public class ProtocolProcessor_CONNECT_Test { // second connect with clean session false m_session = new EmbeddedChannel(); - MqttConnectMessage secondConnMsg = MqttMessageBuilders.connect().clientId(ProtocolProcessorTest.FAKE_CLIENT_ID) + MqttConnectMessage secondConnMsg = MqttMessageBuilders.connect().clientId(FAKE_CLIENT_ID) .protocolVersion(MqttVersion.MQTT_3_1).build(); m_processor.processConnect(m_session, secondConnMsg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); // Verify client session is clean false - ClientSession session = m_sessionStore.sessionForClient(ProtocolProcessorTest.FAKE_CLIENT_ID); + ClientSession session = m_sessionStore.sessionForClient(FAKE_CLIENT_ID); assertFalse(session.isCleanSession()); // Verify @@ -249,16 +215,20 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void connectWithSameClientIDBadCredentialsDoesntDropExistingClient() { // Connect a client1 - MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID) - .username(ProtocolProcessorTest.TEST_USER).password(ProtocolProcessorTest.TEST_PWD).build(); + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) + .username(TEST_USER) + .password(TEST_PWD) + .build(); m_processor.processConnect(m_session, msg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); // create another connect same clientID but with bad credentials - MqttConnectMessage evilClientConnMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1) - .clientId(ProtocolProcessorTest.FAKE_CLIENT_ID) - .username(ProtocolProcessorTest.EVIL_TEST_USER) - .password(ProtocolProcessorTest.EVIL_TEST_PWD).build(); + MqttConnectMessage evilClientConnMsg = MqttMessageBuilders.connect() + .protocolVersion(MqttVersion.MQTT_3_1) + .clientId(FAKE_CLIENT_ID) + .username(ProtocolProcessorTest.EVIL_TEST_USER) + .password(ProtocolProcessorTest.EVIL_TEST_PWD) + .build(); EmbeddedChannel evilSession = new EmbeddedChannel(); @@ -275,9 +245,9 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void testConnAckContainsSessionPresentFlag() throws InterruptedException { - MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID) + MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID) .protocolVersion(MqttVersion.MQTT_3_1_1).build(); - NettyUtils.clientID(m_session, ProtocolProcessorTest.FAKE_CLIENT_ID); + NettyUtils.clientID(m_session, FAKE_CLIENT_ID); NettyUtils.cleanSession(m_session, false); // Connect a first time @@ -297,7 +267,7 @@ public class ProtocolProcessor_CONNECT_Test { @Test public void testMultipleReconnection() throws InterruptedException { // connect with clean a false and subscribe to a topic - MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(ProtocolProcessorTest.FAKE_CLIENT_ID) + MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(FAKE_CLIENT_ID) .protocolVersion(MqttVersion.MQTT_3_1_1).build(); m_processor.processConnect(m_session, msg); assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound()); @@ -307,14 +277,14 @@ public class ProtocolProcessor_CONNECT_Test { MqttSubscribeMessage subscribeMsg = MqttMessageBuilders.subscribe() .addSubscription(MqttQoS.AT_MOST_ONCE, ProtocolProcessorTest.FAKE_TOPIC).messageId(10).build(); - NettyUtils.clientID(m_session, ProtocolProcessorTest.FAKE_CLIENT_ID); + NettyUtils.clientID(m_session, FAKE_CLIENT_ID); NettyUtils.cleanSession(m_session, false); m_processor.processSubscribe(m_session, subscribeMsg); Subscription expectedSubscription = new Subscription( - ProtocolProcessorTest.FAKE_CLIENT_ID, + FAKE_CLIENT_ID, new Topic(ProtocolProcessorTest.FAKE_TOPIC), MqttQoS.AT_MOST_ONCE); - assertTrue(subscriptions.contains(expectedSubscription)); + verifySubscriptionExists(m_session, m_sessionStore, expectedSubscription); assertEqualsSubAck(m_session.readOutbound()); // disconnect @@ -328,22 +298,14 @@ public class ProtocolProcessor_CONNECT_Test { assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen()); // verify that the first subscription is still preserved - assertTrue(subscriptions.contains(expectedSubscription)); + verifySubscriptionExists(m_session, m_sessionStore, expectedSubscription); } @Test public void testZeroByteClientIdWithoutCleanSession() { // Allow zero byte client ids m_processor = new ProtocolProcessor(); - m_processor.init( - subscriptions, - m_messagesStore, - m_sessionStore, - m_mockAuthenticator, - true, - true, - new PermitAllAuthorizator(), - ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients(); // Connect message without clean session set to true but client id is still null MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1) @@ -358,45 +320,40 @@ public class ProtocolProcessor_CONNECT_Test { assertFalse("Connection should be closed by the broker.", m_session.isOpen()); } + protected void reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients() { + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, true, + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + } + @Test public void testZeroByteClientIdWithCleanSession() { // Allow zero byte client ids m_processor = new ProtocolProcessor(); - m_processor.init( - subscriptions, - m_messagesStore, - m_sessionStore, - m_mockAuthenticator, - true, - true, - new PermitAllAuthorizator(), - ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); + reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients(); // Connect message with clean session set to true and client id is null. - MqttConnectMessage msg = connMsg.clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1).cleanSession(true) - .build(); + MqttConnectMessage msg = connMsg.clientId(null) + .protocolVersion(MqttVersion.MQTT_3_1_1) + .cleanSession(true) + .build(); m_processor.processConnect(m_session, msg); - assertEqualsConnAck( - "Connection should be accepted. A unique clientid should be generated" - + " and clean session set to true.", - CONNECTION_ACCEPTED, - m_session.readOutbound()); - + assertEqualsConnAck("Connection MUST be accepted, unique clientid MUST be generated and clean session to true", + CONNECTION_ACCEPTED, m_session.readOutbound()); assertTrue("Connection should be valid and open,", m_session.isOpen()); } @Test public void testZeroByteClientIdNotAllowed() { // Connect message with clean session set to true and client id is null. - MqttConnectMessage msg = connMsg.clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1).cleanSession(true) - .build(); + MqttConnectMessage msg = connMsg.clientId(null) + .protocolVersion(MqttVersion.MQTT_3_1_1) + .cleanSession(true) + .build(); m_processor.processConnect(m_session, msg); - assertEqualsConnAck( - "Zero byte client identifiers are not allowed.", - CONNECTION_REFUSED_IDENTIFIER_REJECTED, - m_session.readOutbound()); + assertEqualsConnAck("Zero byte client identifiers are not allowed", + CONNECTION_REFUSED_IDENTIFIER_REJECTED, m_session.readOutbound()); assertFalse("Connection should closed.", m_session.isOpen()); } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java new file mode 100644 index 00000000..140b89a2 --- /dev/null +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java @@ -0,0 +1,377 @@ +/* + * Copyright (c) 2012-2017 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package io.moquette.spi.impl; + +import io.moquette.server.MessageCollector; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.mqtt.MqttQoS; +import org.eclipse.paho.client.mqttv3.MqttClient; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; + +import static io.netty.handler.codec.mqtt.MqttQoS.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProcessorCommonUtils { + + @Before + public void setUp() throws InterruptedException { + initializeProcessorAndSubsystems(); + } + + @Test + public void testCleanSession_maintainClientSubscriptions() throws InterruptedException { + connectAsClient(this.m_channel, "TestClient"); + subscribe("/topic", AT_MOST_ONCE); + disconnect(); + + // reconnect and publish + EmbeddedChannel anotherChannel = new EmbeddedChannel(); +// NettyUtils.clientID(anotherChannel, "TestClient"); +// NettyUtils.cleanSession(anotherChannel, false); + connectAsClient(anotherChannel, "TestClient"); + + publishToAs(anotherChannel, "TestClient", "/topic", AT_MOST_ONCE, false); + + verifyPublishIsReceived(anotherChannel); + } + + @Ignore("At the end it's equal to testCleanSession_maintainClientSubscriptions") + @Test + public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() throws InterruptedException { + connectAsClient("TestClient"); + subscribe("/topic", AT_MOST_ONCE); + disconnect(); + + this.m_channel = new EmbeddedChannel(); + // reconnect and publish + connectAsClient("TestClient"); + + publishToAs("TestClient", "/topic", AT_MOST_ONCE, false); + + verifyPublishIsReceived(); + } + + /** + * Check that after a client has connected with clean session false, subscribed to some topic + * and exited, if it reconnects with clean session true, the broker correctly cleanup every + * previous subscription + */ + @Test + public void testCleanSession_correctlyClientSubscriptions() throws InterruptedException { + connectAsClient("TestClient"); + subscribe("/topic", AT_MOST_ONCE); + disconnect(); + + this.m_channel = new EmbeddedChannel(); + // reconnect and publish + connectWithCleanSession("TestClient"); + + publishToAs("TestClient", "/topic", AT_MOST_ONCE, false); + + verifyNoPublishIsReceived(); + } + + @Test + public void testRetain_maintainMessage_againstClientDestruction() throws InterruptedException { + connect(); + publishToAs(FAKE_CLIENT_ID, "/topic", AT_LEAST_ONCE, 66, true); + disconnect(); + + // reconnect and publish + this.m_channel = new EmbeddedChannel(); + connect(); + subscribe("/topic", AT_MOST_ONCE); + + verifyPublishIsReceived(); + } + + @Test + public void testUnsubscribe_do_not_notify_anymore_same_session() { + connect(); + subscribe("/topic", AT_MOST_ONCE); + publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, false); + verifyPublishIsReceived(); + + unsubscribeAndVerifyAck("/topic"); + publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, false); + verifyNoPublishIsReceived(); + } + + @Test + public void testUnsubscribe_do_not_notify_anymore_new_session() throws InterruptedException { + connect(); + subscribe("/topic", AT_MOST_ONCE); + publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, false); + verifyPublishIsReceived(); + + unsubscribeAndVerifyAck("/topic"); + disconnect(); + + this.m_channel = new EmbeddedChannel(); + connect(); + publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, false); + verifyNoPublishIsReceived(); + } + + @Test + public void testPublishWithQoS1() throws InterruptedException { + connect(); + subscribe("/topic", AT_LEAST_ONCE); + publishToAs(FAKE_CLIENT_ID, "/topic", AT_LEAST_ONCE, 66, false); + disconnect(); + + verifyPublishIsReceived(AT_LEAST_ONCE); + } + + @Test + public void testPublishWithQoS1_notCleanSession() throws InterruptedException { + connect(); + subscribe("/topic", AT_LEAST_ONCE); + disconnect(); + + // publish a QoS 1 message another client publish a message on the topic + publishFromAnotherClient("/topic", AT_LEAST_ONCE); + + this.m_channel = new EmbeddedChannel(); + connect(); + verifyPublishIsReceived(); + } + + protected void publishFromAnotherClient(String topic, MqttQoS qos) throws InterruptedException { + EmbeddedChannel anotherChannel = new EmbeddedChannel(); + connectAsClient(anotherChannel, "AnotherClient"); + publishToAs(anotherChannel, "AnotherClient", topic, qos, 67, false); + disconnect(anotherChannel); + } + + protected void publishFromAnotherClient(String topic, String payload, MqttQoS qos) throws InterruptedException { + EmbeddedChannel anotherChannel = new EmbeddedChannel(); + connectAsClient(anotherChannel, "AnotherClient"); + publishToAs(anotherChannel, "AnotherClient", topic, payload, qos, 68, false); + disconnect(anotherChannel); + } + + protected void publishQoS2FromAnotherClient(String topic) throws InterruptedException { + EmbeddedChannel anotherChannel = new EmbeddedChannel(); + connectAsClient(anotherChannel, "AnotherClient"); + publishQoS2ToAs(anotherChannel, "AnotherClient", topic, 67, false); + disconnect(anotherChannel); + } + + @Test + public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws InterruptedException { + connect(); + subscribe("/topic", AT_LEAST_ONCE); + disconnect(); + + this.m_channel = new EmbeddedChannel(); + connect(); + subscribe("/topic", AT_LEAST_ONCE); + + // publish a QoS 1 message another client publish a message on the topic + publishFromAnotherClient("/topic", AT_LEAST_ONCE); + + // Verify that after a reconnection the client receive the message + verifyPublishIsReceived(); + } + + @Test + public void testPublishWithQoS2() throws InterruptedException { + connect(); + subscribe("/topic", EXACTLY_ONCE); + disconnect(); + + publishQoS2FromAnotherClient("/topic"); + + this.m_channel = new EmbeddedChannel(); + connect(); + + verifyPublishIsReceived(EXACTLY_ONCE); + } + + @Test + public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() throws InterruptedException { + connect(); + subscribe("/topic", AT_LEAST_ONCE); + disconnect(); + + publishFromAnotherClient("/topic", "Hello MQTT 1", AT_LEAST_ONCE); + + this.m_channel = new EmbeddedChannel(); + connect(); + verifyPublishIsReceived("Hello MQTT 1", AT_LEAST_ONCE); + disconnect(); + + // publish other message + publishFromAnotherClient("/topic", "Hello MQTT 2", AT_LEAST_ONCE); + + this.m_channel = new EmbeddedChannel(); + connect(); + verifyPublishIsReceived("Hello MQTT 2", AT_LEAST_ONCE); + } + + + @Test + public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws InterruptedException { + connect(); + subscribe("/topic", AT_MOST_ONCE); + publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, true); + disconnect(); + + // second loop + this.m_channel = new EmbeddedChannel(); + connect(); + subscribe("/topic", AT_MOST_ONCE); + publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, true); + disconnect(); + + assertFalse("after a disconnect the client should be disconnected", this.m_channel.isOpen()); + } + + @Test + public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws InterruptedException { + EmbeddedChannel publisher = new EmbeddedChannel(); + connectAsClient(publisher, "Publisher"); + + connect(); + subscribe("/topic", AT_LEAST_ONCE); + + publishToAs(publisher, "Publisher", "/topic", "Hello world MQTT!!-1", AT_LEAST_ONCE, 99, false); + verifyPublishIsReceived("Hello world MQTT!!-1", AT_LEAST_ONCE); + disconnect(); + + publishToAs(publisher, "Publisher", "/topic", "Hello world MQTT!!-2", AT_LEAST_ONCE, 100, false); + publishToAs(publisher, "Publisher", "/topic", "Hello world MQTT!!-3", AT_LEAST_ONCE, 101, false); + + this.m_channel = new EmbeddedChannel(); + connect(); + verifyPublishIsReceived("Hello world MQTT!!-2", AT_LEAST_ONCE); + verifyPublishIsReceived("Hello world MQTT!!-3", AT_LEAST_ONCE); + } + + @Test + public void checkReplayStoredPublish_forNoCleanSession_qos1() throws InterruptedException { + EmbeddedChannel publisher = new EmbeddedChannel(); + connectAsClient(publisher, "Publisher"); + + connect(); + subscribe("/topic", AT_LEAST_ONCE); + disconnect(); + + publishToAs(publisher, "Publisher", "/topic", "Hello world MQTT!!-1", AT_LEAST_ONCE, 99, false); + + this.m_channel = new EmbeddedChannel(); + connect(); + subscribeAndNotReadResponse("/topic", AT_LEAST_ONCE); + verifyPublishIsReceived("Hello world MQTT!!-1", AT_LEAST_ONCE); + } + + /** + * subscriber connect and subscribe on "topic" subscriber disconnects publisher connects and + * send two message "hello1" "hello2" to "topic" subscriber connects again and receive "hello1" + * "hello2" + */ + @Test + public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() throws InterruptedException { + connect(); + subscribe("topic", EXACTLY_ONCE); + disconnect(); + + EmbeddedChannel publisher = new EmbeddedChannel(); + connectAsClient(publisher, "Publisher"); + publishQoS2ToAs(publisher, "Publisher", "topic", "Hello1", 101, true); + publishQoS2ToAs(publisher, "Publisher", "topic", "Hello2", 102, true); + disconnect(publisher); + + this.m_channel = new EmbeddedChannel(); + connect(); + verifyPublishIsReceived("Hello1", EXACTLY_ONCE); + verifyPublishIsReceived("Hello2", EXACTLY_ONCE); + } + + /** + * subscriber connect and subscribe on "a/b" QoS 1 and "a/+" QoS 2 publisher connects and send a + * message "hello" on "a/b" subscriber must receive only a single message not twice + */ + @Test + public void checkSinglePublishOnOverlappingSubscriptions() { + EmbeddedChannel publisher = new EmbeddedChannel(); + connectAsClient(publisher, "Publisher"); + + connect(); + subscribe("a/+", EXACTLY_ONCE); + subscribe("a/b", AT_LEAST_ONCE); + + // force the publisher to send + publishToAs(publisher, "Publisher", "a/b", "Hello world MQTT!!", AT_LEAST_ONCE, 60, false); + + verifyPublishIsReceived("Hello world MQTT!!", AT_LEAST_ONCE); + // try to listen a second publish + verifyNoPublishIsReceived(); + } + + @Test + public void testForceClientDisconnection_issue116() { + EmbeddedChannel clientXA = new EmbeddedChannel(); + connectAsClient(clientXA, "subscriber"); + subscribe(clientXA, "topic", AT_MOST_ONCE); + + EmbeddedChannel clientXB = new EmbeddedChannel(); + connectAsClient(clientXB, "publisher"); + publishQoS2ToAs(clientXB, "publisher", "topic", "Hello", 20, false); + + EmbeddedChannel clientYA = new EmbeddedChannel(); + connectAsClient(clientYA, "subscriber"); + subscribe(clientYA, "topic", AT_MOST_ONCE); + + EmbeddedChannel clientYB = new EmbeddedChannel(); + connectAsClient(clientYB, "publisher"); + publishQoS2ToAs(clientYB, "publisher", "topic", "Hello 2", 20, true); + + assertFalse("First 'subscriber' channel MUST be closed by the broker", clientXA.isOpen()); + verifyPublishIsReceived(clientYA, "Hello 2", AT_MOST_ONCE); + +// LOG.info("*** testForceClientDisconnection_issue118 ***"); +// MessageCollector cbSubscriber1 = new MessageCollector(); +// MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1); +// LOG.info("Connected 'subscriber' first time"); +// clientXA.subscribe("topic", 0); +// LOG.info("Subscribed 'topic' from 'subscriber' first time"); +// +// MqttClient clientXB = createClient("publisher", "X"); +// LOG.info("Connected 'publisher' first time"); +// clientXB.publish("topic", "Hello".getBytes(), 2, false); +// LOG.info("Published on 'topic' from 'publisher' first time"); +// +// LOG.info("Creating second new 'subscriber'"); +// MessageCollector cbSubscriber2 = new MessageCollector(); +// MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2); +// LOG.info("Connected 'subscriber' second time"); +// clientYA.subscribe("topic", 0); +// +// MqttClient clientYB = createClient("publisher", "Y"); +// clientYB.publish("topic", "Hello 2".getBytes(), 2, true); +// +// // Verify that the second subscriber client get notified and not the first. +// assertTrue(cbSubscriber1.connectionLost()); +// assertEquals("Hello 2", new String(cbSubscriber2.waitMessage(1).getPayload())); + } + +}