From efbfab791493fb7bdc2e2b943abdac326c147cb5 Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 23 Feb 2016 23:29:45 +0100 Subject: [PATCH] Minor, moved tests from ServerIntegrationFuseTest to ServerIntegrationPahoTest to limit the usage of fusesource mqtt client --- .../io/moquette/server/MessageCollector.java | 98 ++++++++++ .../ServerIntegrationEmbeddedPublishTest.java | 75 +++++--- .../server/ServerIntegrationFuseTest.java | 176 +----------------- .../server/ServerIntegrationPahoTest.java | 141 +++++++++++++- 4 files changed, 283 insertions(+), 207 deletions(-) create mode 100644 broker/src/test/java/io/moquette/server/MessageCollector.java diff --git a/broker/src/test/java/io/moquette/server/MessageCollector.java b/broker/src/test/java/io/moquette/server/MessageCollector.java new file mode 100644 index 00000000..0ea8b43c --- /dev/null +++ b/broker/src/test/java/io/moquette/server/MessageCollector.java @@ -0,0 +1,98 @@ +/* + * Copyright (c) 2012-2016 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.server; + +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; +import org.eclipse.paho.client.mqttv3.MqttCallback; +import org.eclipse.paho.client.mqttv3.MqttMessage; + +import java.util.concurrent.*; + +/** + * Used in test to collect all messages received asynchronously by MqttClient. + */ +public class MessageCollector implements MqttCallback { + + private class ReceivedMessage { + private final MqttMessage message; + private final String topic; + + private ReceivedMessage(MqttMessage message, String topic) { + this.message = message; + this.topic = topic; + } + } + + private BlockingQueue m_messages = new LinkedBlockingQueue<>(); + private boolean m_connectionLost = false; + + public MqttMessage getMessage(boolean checkElapsed) { + if (!checkElapsed && m_messages.isEmpty()) { + return null; + } + try { + return m_messages.poll(1, TimeUnit.SECONDS).message; + } catch (InterruptedException e) { + return null; + } + } + + public MqttMessage getMessage(int delay) { + try { + ReceivedMessage msg = m_messages.poll(delay, TimeUnit.SECONDS); + if (msg == null) { + return null; + } + return msg.message; + } catch (InterruptedException e) { + return null; + } + } + + public String getTopic() { + try { + return m_messages.poll(5, TimeUnit.SECONDS).topic; + } catch (InterruptedException e) { + return null; + } + } + + void reinit() { + m_messages = new LinkedBlockingQueue<>(); + m_connectionLost = false; + } + + public boolean connectionLost() { + return m_connectionLost; + } + + @Override + public void connectionLost(Throwable cause) { + System.out.println("connectionLost"); + m_connectionLost = true; + } + + @Override + public void messageArrived(String topic, MqttMessage message) throws Exception { + System.out.println("Message received"); + m_messages.offer(new ReceivedMessage(message, topic)); + } + + @Override + public void deliveryComplete(IMqttDeliveryToken token) { + System.out.println("deliveryComplete"); + } +} diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java index 5648496c..389d9519 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java @@ -19,18 +19,19 @@ import io.moquette.parser.proto.messages.AbstractMessage.QOSType; import io.moquette.parser.proto.messages.PublishMessage; import io.moquette.server.config.IConfig; import io.moquette.server.config.MemoryConfig; -import org.fusesource.mqtt.client.*; +import org.eclipse.paho.client.mqttv3.*; +import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; import org.junit.After; import org.junit.Before; +import org.junit.BeforeClass; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.io.EOFException; +import java.io.File; import java.io.IOException; import java.nio.ByteBuffer; import java.util.Properties; -import java.util.concurrent.TimeUnit; import static org.junit.Assert.*; @@ -38,9 +39,12 @@ public class ServerIntegrationEmbeddedPublishTest { private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationEmbeddedPublishTest.class); + static MqttClientPersistence s_dataStore; + static MqttClientPersistence s_pubDataStore; + Server m_server; - MQTT m_mqtt; - BlockingConnection m_subscriber; + IMqttClient m_subscriber; + TestCallback m_callback; IConfig m_config; protected void startServer() throws IOException { @@ -50,18 +54,28 @@ public class ServerIntegrationEmbeddedPublishTest { m_server.startServer(m_config); } + @BeforeClass + public static void beforeTests() { + String tmpDir = System.getProperty("java.io.tmpdir"); + s_dataStore = new MqttDefaultFilePersistence(tmpDir); + s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher"); + } + @Before public void setUp() throws Exception { + String dbPath = IntegrationUtils.localMapDBPath(); + IntegrationUtils.cleanPersistenceFile(dbPath); + startServer(); - m_mqtt = new MQTT(); - m_mqtt.setHost("localhost", 1883); - m_mqtt.setClientId("Subscriber"); + m_subscriber = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); + m_callback = new TestCallback(); + m_subscriber.setCallback(m_callback); } @After public void tearDown() throws Exception { - if (m_subscriber != null) { + if (m_subscriber != null && m_subscriber.isConnected()) { m_subscriber.disconnect(); } @@ -69,11 +83,9 @@ public class ServerIntegrationEmbeddedPublishTest { IntegrationUtils.cleanPersistenceFile(m_config); } - private void subscribeToWithQos(String topic, QoS qos) throws Exception { - m_subscriber = m_mqtt.blockingConnection(); + private void subscribeToWithQos(String topic, int qos) throws Exception { m_subscriber.connect(); - Topic[] topics = new Topic[]{new Topic(topic, qos)}; - m_subscriber.subscribe(topics); + m_subscriber.subscribe(topic, qos); } private void internalPublishToWithQosAndRetained(String topic, QOSType qos, boolean retained) { @@ -86,13 +98,13 @@ public class ServerIntegrationEmbeddedPublishTest { } private void verifyNoMessageIsReceived() throws Exception { - Message msg = m_subscriber.receive(1, TimeUnit.SECONDS); + MqttMessage msg = m_callback.getMessage(false); assertNull(msg); } private void verifyMessageIsReceivedSuccessfully() throws Exception { - Message msg = m_subscriber.receive(2, TimeUnit.SECONDS); - msg.ack(); + //check in 2 seconds + MqttMessage msg = m_callback.getMessage(2); assertNotNull(msg); assertEquals("Hello world MQTT!!", new String(msg.getPayload())); } @@ -103,7 +115,7 @@ public class ServerIntegrationEmbeddedPublishTest { //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.MOST_ONE, false); - subscribeToWithQos("/topic", QoS.AT_MOST_ONCE); + subscribeToWithQos("/topic", 0); //Verify verifyNoMessageIsReceived(); @@ -113,7 +125,7 @@ public class ServerIntegrationEmbeddedPublishTest { public void testClientSubscribeBeforeNotRetainedQoS0IsSent() throws Exception { LOG.info("*** testClientSubscribeBeforeNotRetainedQoS0 ***"); - subscribeToWithQos("/topic", QoS.AT_MOST_ONCE); + subscribeToWithQos("/topic", 0); //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.MOST_ONE, false); @@ -126,7 +138,7 @@ public class ServerIntegrationEmbeddedPublishTest { public void testClientSubscribeBeforeRetainedQoS0IsSent() throws Exception { LOG.info("*** testClientSubscribeBeforeRetainedQoS0IsSent ***"); - subscribeToWithQos("/topic", QoS.AT_MOST_ONCE); + subscribeToWithQos("/topic", 0); //super ugly but we need the MQTT client lib finish it's job before us Thread.sleep(1000); @@ -143,7 +155,7 @@ public class ServerIntegrationEmbeddedPublishTest { //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.MOST_ONE, true); - subscribeToWithQos("/topic", QoS.AT_MOST_ONCE); + subscribeToWithQos("/topic", 0); //Verify verifyNoMessageIsReceived(); @@ -153,7 +165,7 @@ public class ServerIntegrationEmbeddedPublishTest { public void testClientSubscribeBeforeNotRetainedQoS1IsSent() throws Exception { LOG.info("*** testClientSubscribeBeforeNotRetainedQoS1IsSent ***"); - subscribeToWithQos("/topic", QoS.AT_LEAST_ONCE); + subscribeToWithQos("/topic", 1); //super ugly but we need the MQTT client lib finish it's job before us Thread.sleep(1000); @@ -170,7 +182,7 @@ public class ServerIntegrationEmbeddedPublishTest { //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.LEAST_ONE, false); - subscribeToWithQos("/topic", QoS.AT_LEAST_ONCE); + subscribeToWithQos("/topic", 1); //Verify verifyNoMessageIsReceived(); @@ -180,7 +192,7 @@ public class ServerIntegrationEmbeddedPublishTest { public void testClientSubscribeBeforeRetainedQoS1IsSent() throws Exception { LOG.info("*** testClientSubscribeBeforeRetainedQoS1IsSent ***"); - subscribeToWithQos("/topic", QoS.AT_LEAST_ONCE); + subscribeToWithQos("/topic", 1); //super ugly but we need the MQTT client lib finish it's job before us Thread.sleep(1000); @@ -197,7 +209,7 @@ public class ServerIntegrationEmbeddedPublishTest { //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.LEAST_ONE, true); - subscribeToWithQos("/topic", QoS.AT_LEAST_ONCE); + subscribeToWithQos("/topic", 1); //Verify verifyMessageIsReceivedSuccessfully(); @@ -207,7 +219,7 @@ public class ServerIntegrationEmbeddedPublishTest { public void testClientSubscribeBeforeNotRetainedQoS2IsSent() throws Exception { LOG.info("*** testClientSubscribeBeforeNotRetainedQoS2IsSent ***"); - subscribeToWithQos("/topic", QoS.EXACTLY_ONCE); + subscribeToWithQos("/topic", 2); //super ugly but we need the MQTT client lib finish it's job before us Thread.sleep(1000); @@ -224,7 +236,7 @@ public class ServerIntegrationEmbeddedPublishTest { //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.EXACTLY_ONCE, false); - subscribeToWithQos("/topic", QoS.EXACTLY_ONCE); + subscribeToWithQos("/topic", 2); //Verify verifyNoMessageIsReceived(); @@ -234,7 +246,7 @@ public class ServerIntegrationEmbeddedPublishTest { public void testClientSubscribeBeforeRetainedQoS2IsSent() throws Exception { LOG.info("*** testClientSubscribeBeforeRetainedQoS2IsSent ***"); - subscribeToWithQos("/topic", QoS.EXACTLY_ONCE); + subscribeToWithQos("/topic", 2); //super ugly but we need the MQTT client lib finish it's job before us Thread.sleep(1000); @@ -251,22 +263,23 @@ public class ServerIntegrationEmbeddedPublishTest { //Exercise internalPublishToWithQosAndRetained("/topic", QOSType.EXACTLY_ONCE, true); - subscribeToWithQos("/topic", QoS.EXACTLY_ONCE); + subscribeToWithQos("/topic", 2); //Verify verifyMessageIsReceivedSuccessfully(); } - @Test(expected = EOFException.class) + @Test(expected = IllegalStateException.class) public void testClientSubscribeAfterDisconnected() throws Exception { LOG.info("*** testClientSubscribeAfterDisconnected ***"); - subscribeToWithQos("foo", QoS.AT_MOST_ONCE); + subscribeToWithQos("foo", 0); m_subscriber.disconnect(); internalPublishToWithQosAndRetained("foo", QOSType.MOST_ONE, false); //verifyMessageIsReceivedSuccessfully(); - m_subscriber.receive(2, TimeUnit.MILLISECONDS); + //m_subscriber.receive(2, TimeUnit.MILLISECONDS); + m_callback.getMessage(true); } } diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java index 6eac6601..8faaef59 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationFuseTest.java @@ -71,22 +71,7 @@ public class ServerIntegrationFuseTest { m_server.stopServer(); IntegrationUtils.cleanPersistenceFile(m_config); } - - @Test - public void connectWithCredentials() throws Exception { - LOG.info("*** connectWithCredetials ***"); - MQTT mqtt = new MQTT(); - mqtt.setHost("localhost", 1883); - mqtt.setClientId("Publisher"); - mqtt.setUserName("testuser"); - mqtt.setPassword("passwd"); - - m_publisher = mqtt.blockingConnection(); - m_publisher.connect(); - - assertTrue(true); - } - + @Test public void checkWillTestamentIsPublishedOnConnectionKill_noRetain() throws Exception { LOG.info("checkWillTestamentIsPublishedOnConnectionKill"); @@ -121,163 +106,4 @@ public class ServerIntegrationFuseTest { msg.ack(); assertEquals(willTestamentMsg, new String(msg.getPayload())); } - - @Test - public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws Exception { - LOG.info("*** checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1 ***"); - MQTT mqtt = new MQTT(); - mqtt.setHost("localhost", 1883); - mqtt.setClientId("Publisher"); - m_publisher = mqtt.blockingConnection(); - m_publisher.connect(); - - m_mqtt.setHost("localhost", 1883); - m_mqtt.setCleanSession(false); - m_mqtt.setClientId("Subscriber"); - m_subscriber = m_mqtt.blockingConnection(); - m_subscriber.connect(); - Topic[] topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)}; - m_subscriber.subscribe(topics); - - //force the publisher to send - m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), QoS.AT_LEAST_ONCE, false); - - //read the first message and drop the connection - Message msg = m_subscriber.receive(); - msg.ack(); - assertEquals("Hello world MQTT!!-1", new String(msg.getPayload())); - m_subscriber.disconnect(); - - m_publisher.publish("/topic", "Hello world MQTT!!-2".getBytes(), QoS.AT_LEAST_ONCE, false); - m_publisher.publish("/topic", "Hello world MQTT!!-3".getBytes(), QoS.AT_LEAST_ONCE, false); - - //reconnect and expect to receive the hello 2 message - m_subscriber = m_mqtt.blockingConnection(); - m_subscriber.connect(); - //topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)}; - //m_subscriber.subscribe(topics); - msg = m_subscriber.receive(); - msg.ack(); - assertEquals("Hello world MQTT!!-2", new String(msg.getPayload())); - - msg = m_subscriber.receive(); - assertEquals("Hello world MQTT!!-3", new String(msg.getPayload())); - msg.ack(); - m_subscriber.disconnect(); - } - - @Test - public void checkReplayStoredPublish_forNoCleanSession_qos1() throws Exception { - LOG.info("*** checkReplayStoredPublish_forNoCleanSession_qos1 ***"); - MQTT mqtt = new MQTT(); - mqtt.setHost("localhost", 1883); - mqtt.setClientId("Publisher"); - m_publisher = mqtt.blockingConnection(); - m_publisher.connect(); - - m_mqtt.setHost("localhost", 1883); - m_mqtt.setCleanSession(false); - m_mqtt.setClientId("Subscriber"); - m_subscriber = m_mqtt.blockingConnection(); - m_subscriber.connect(); - Topic[] topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)}; - m_subscriber.subscribe(topics); - m_subscriber.disconnect(); - - //force the publisher to send - m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), QoS.AT_LEAST_ONCE, false); - - //reconnect and expect to receive the hello 2 message - m_subscriber = m_mqtt.blockingConnection(); - m_subscriber.connect(); - topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)}; - m_subscriber.subscribe(topics); - Message msg = m_subscriber.receive(1, TimeUnit.SECONDS); - assertNotNull(msg); - msg.ack(); - assertEquals("Hello world MQTT!!-1", new String(msg.getPayload())); -// m_subscriber.disconnect(); - } - - /** - * 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 ***"); - m_mqtt.setHost("localhost", 1883); - m_mqtt.setCleanSession(false); - m_mqtt.setClientId("Subscriber"); - m_subscriber = m_mqtt.blockingConnection(); - m_subscriber.connect(); - Topic[] topics = new Topic[]{new Topic("topic", QoS.EXACTLY_ONCE)}; - m_subscriber.subscribe(topics); - m_subscriber.disconnect(); - - MQTT mqtt = new MQTT(); - mqtt.setHost("localhost", 1883); - mqtt.setClientId("Publisher"); - m_publisher = mqtt.blockingConnection(); - m_publisher.connect(); - m_publisher.publish("topic", "Hello1".getBytes(), QoS.EXACTLY_ONCE, true); - m_publisher.publish("topic", "Hello2".getBytes(), QoS.EXACTLY_ONCE, true); - m_publisher.disconnect(); - - //subscriber reconnects - m_mqtt = new MQTT(); - m_mqtt.setHost("localhost", 1883); - m_mqtt.setClientId("Subscriber"); - m_mqtt.setCleanSession(false); - m_subscriber = m_mqtt.blockingConnection(); - m_subscriber.connect(); - //subscriber should receive the 2 messages missed - Message msg = m_subscriber.receive(1, TimeUnit.SECONDS); - assertNotNull(msg); - msg.ack(); - assertEquals("Hello1", new String(msg.getPayload())); - msg = m_subscriber.receive(1, TimeUnit.SECONDS); - assertNotNull(msg); - msg.ack(); - 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 ***"); - MQTT mqtt = new MQTT(); - mqtt.setHost("localhost", 1883); - mqtt.setClientId("Publisher"); - m_publisher = mqtt.blockingConnection(); - m_publisher.connect(); - - m_mqtt.setHost("localhost", 1883); - m_mqtt.setCleanSession(false); - m_mqtt.setClientId("Subscriber"); - m_subscriber = m_mqtt.blockingConnection(); - m_subscriber.connect(); - Topic[] topics = new Topic[]{ - new Topic("a/+", QoS.EXACTLY_ONCE), - new Topic("a/b", QoS.AT_LEAST_ONCE)}; - m_subscriber.subscribe(topics); - - //force the publisher to send - m_publisher.publish("a/b", "Hello world MQTT!!".getBytes(), QoS.AT_LEAST_ONCE, false); - - //reconnect and expect to receive the hello 2 message - Message msg = m_subscriber.receive(1, TimeUnit.SECONDS); - assertNotNull(msg); - msg.ack(); - assertEquals("Hello world MQTT!!", new String(msg.getPayload())); - //try to listen a second publish - msg = m_subscriber.receive(1, TimeUnit.SECONDS); - assertNull(msg); - } } diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index 0676b25a..01033ab2 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java @@ -41,6 +41,7 @@ public class ServerIntegrationPahoTest { Server m_server; IMqttClient m_client; + IMqttClient m_publisher; TestCallback m_callback; IConfig m_config; @@ -68,14 +69,20 @@ public class ServerIntegrationPahoTest { m_client = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); m_callback = new TestCallback(); m_client.setCallback(m_callback); + + m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); } @After public void tearDown() throws Exception { - if (m_client.isConnected()) { + if (m_client != null && m_client.isConnected()) { m_client.disconnect(); } + if (m_publisher != null && m_publisher.isConnected()) { + m_publisher.disconnect(); + } + stopServer(); } @@ -84,6 +91,18 @@ public class ServerIntegrationPahoTest { IntegrationUtils.cleanPersistenceFile(m_config); } + @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 ***"); @@ -484,6 +503,126 @@ public class ServerIntegrationPahoTest { 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 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); + + //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 = messagesCollector.getMessage(true); + 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 = messagesCollector.getMessage(true); + assertEquals("Hello world MQTT!!-2", new String(msg.getPayload())); + + msg = messagesCollector.getMessage(true); + 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.getMessage(true); + 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.getMessage(1); + assertNotNull(msg); + assertEquals("Hello1", new String(msg.getPayload())); + msg = messagesCollector.getMessage(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.getMessage(1); + assertEquals("Hello world MQTT!!", new String(msg.getPayload())); + + //try to listen a second publish + msg = messagesCollector.getMessage(1); + assertNull(msg); + } + protected MqttClient createClient(String clientName, String storeSuffix) throws MqttException { return createClient(clientName, storeSuffix, null);