diff --git a/broker/src/test/java/io/moquette/server/MessageCollector.java b/broker/src/test/java/io/moquette/server/MessageCollector.java index 0ea8b43c..c93c6935 100644 --- a/broker/src/test/java/io/moquette/server/MessageCollector.java +++ b/broker/src/test/java/io/moquette/server/MessageCollector.java @@ -81,18 +81,15 @@ public class MessageCollector implements MqttCallback { @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/ServerLowlevelMessagesIntegrationTests.java b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java index 08b10767..8dd17ef7 100644 --- a/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java +++ b/broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java @@ -22,7 +22,10 @@ import io.moquette.parser.proto.messages.ConnAckMessage; import io.moquette.server.config.IConfig; import io.moquette.server.config.MemoryConfig; import io.moquette.testclient.Client; -import org.fusesource.mqtt.client.*; +import org.eclipse.paho.client.mqttv3.IMqttClient; +import org.eclipse.paho.client.mqttv3.MqttClient; +import org.eclipse.paho.client.mqttv3.MqttClientPersistence; +import org.eclipse.paho.client.mqttv3.MqttMessage; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -42,10 +45,11 @@ import static org.junit.Assert.*; */ public class ServerLowlevelMessagesIntegrationTests { private static final Logger LOG = LoggerFactory.getLogger(ServerLowlevelMessagesIntegrationTests.class); - + static MqttClientPersistence s_dataStore; Server m_server; Client m_client; - MQTT m_subscriberDef; + IMqttClient m_willSubscriber; + MessageCollector m_messageCollector; IConfig m_config; protected void startServer() throws IOException { @@ -59,9 +63,9 @@ public class ServerLowlevelMessagesIntegrationTests { public void setUp() throws Exception { startServer(); m_client = new Client("localhost"); - m_subscriberDef = new MQTT(); - m_subscriberDef.setHost("localhost", 1883); - m_subscriberDef.setClientId("Subscriber"); + m_willSubscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_dataStore); + m_messageCollector = new MessageCollector(); + m_willSubscriber.setCallback(m_messageCollector); } @After @@ -95,10 +99,8 @@ public class ServerLowlevelMessagesIntegrationTests { String willTestamentTopic = "/will/test"; String willTestamentMsg = "Bye bye"; - BlockingConnection willSubscriber = m_subscriberDef.blockingConnection(); - willSubscriber.connect(); - Topic[] topics = new Topic[]{new Topic(willTestamentTopic, QoS.AT_MOST_ONCE)}; - willSubscriber.subscribe(topics); + m_willSubscriber.connect(); + m_willSubscriber.subscribe(willTestamentTopic, 0); int keepAlive = 2; //secs ConnectMessage connectMessage = new ConnectMessage(); @@ -116,18 +118,17 @@ public class ServerLowlevelMessagesIntegrationTests { //but after the 2 KEEP ALIVE timeout expires it gets fired, //NB it's 1,5 * KEEP_ALIVE so 3 secs and some millis to propagate the message - Message msg = willSubscriber.receive(3300, TimeUnit.MILLISECONDS); + MqttMessage msg = m_messageCollector.getMessage(3300); long willMessageReceiveTime = System.currentTimeMillis(); if (msg == null) { LOG.warn("testament message is null"); } assertNotNull("the will message should be fired after keep alive!", msg); - msg.ack(); //the will message hasn't to be received before the elapsing of Keep Alive timeout assertTrue(willMessageReceiveTime - connectTime > 3000); assertEquals(willTestamentMsg, new String(msg.getPayload())); - willSubscriber.disconnect(); + m_willSubscriber.disconnect(); } AbstractMessage receivedMsg; diff --git a/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java b/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java index fab59bdb..b693afd4 100644 --- a/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java +++ b/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java @@ -17,13 +17,16 @@ package io.moquette.server; 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 java.io.File; import java.io.IOException; import java.util.Properties; -import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; @@ -33,12 +36,16 @@ import static org.junit.Assert.assertNull; */ public class ServerRestartIntegrationTest { + static MqttClientPersistence s_dataStore; + static MqttClientPersistence s_pubDataStore; + static MqttConnectOptions CLEAN_SESSION_OPT = new MqttConnectOptions(); + Server m_server; - MQTT m_mqtt; - BlockingConnection m_subscriber; - BlockingConnection m_publisher; + IMqttClient m_subscriber; + IMqttClient m_publisher; IConfig m_config; - + MessageCollector m_messageCollector; + protected void startServer() throws IOException { m_server = new Server(); final Properties configProps = IntegrationUtils.prepareTestPropeties(); @@ -46,23 +53,34 @@ public class ServerRestartIntegrationTest { 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"); + CLEAN_SESSION_OPT.setCleanSession(false); + } + @Before public void setUp() throws Exception { startServer(); + String dbPath = IntegrationUtils.localMapDBPath(); + IntegrationUtils.cleanPersistenceFile(dbPath); - m_mqtt = new MQTT(); + m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_dataStore); + m_messageCollector = new MessageCollector(); + m_subscriber.setCallback(m_messageCollector); - m_mqtt.setHost("localhost", 1883); - m_mqtt.setCleanSession(false); + m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); } @After public void tearDown() throws Exception { - if (m_subscriber != null) { + if (m_subscriber != null && m_subscriber.isConnected()) { m_subscriber.disconnect(); } - - if (m_publisher != null) { + + if (m_publisher != null && m_publisher.isConnected()) { m_publisher.disconnect(); } @@ -73,12 +91,9 @@ public class ServerRestartIntegrationTest { @Test public void checkRestartCleanSubscriptionTree() throws Exception { - m_mqtt.setClientId("Subscriber"); //subscribe to /topic - 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.connect(CLEAN_SESSION_OPT); + m_subscriber.subscribe("/topic", 1); m_subscriber.disconnect(); //shutdown the server @@ -89,32 +104,24 @@ public class ServerRestartIntegrationTest { m_server.startServer(IntegrationUtils.prepareTestPropeties()); //reconnect the Subscriber subscribing to the same /topic but different QoS - m_subscriber = m_mqtt.blockingConnection(); - m_subscriber.connect(); - topics = new Topic[]{new Topic("/topic", QoS.EXACTLY_ONCE)}; - m_subscriber.subscribe(topics); + m_subscriber.connect(CLEAN_SESSION_OPT); + m_subscriber.subscribe("/topic", 2); //should be just one registration so a publisher receive one notification - MQTT mqtt = new MQTT(); - mqtt.setHost("localhost", 1883); - mqtt.setClientId("Publisher"); - m_publisher = mqtt.blockingConnection(); - m_publisher.connect(); - m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), QoS.AT_LEAST_ONCE, false); + m_publisher.connect(CLEAN_SESSION_OPT); + m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), 1, false); //read the messages - Message msg = m_subscriber.receive(); - msg.ack(); + MqttMessage msg = m_messageCollector.getMessage(true); assertEquals("Hello world MQTT!!", new String(msg.getPayload())); //no more messages on the same topic will be received - assertNull(m_subscriber.receive(1, TimeUnit.SECONDS)); + assertNull(m_messageCollector.getMessage(1)); } @Test public void checkDontPublishInactiveClientsAfterServerRestart() throws Exception { - m_mqtt.setClientId("SubPub"); - BlockingConnection conn = subscribeAndPublish("/topic"); + IMqttClient conn = subscribeAndPublish("/topic"); conn.disconnect(); //shutdown the server @@ -123,25 +130,18 @@ public class ServerRestartIntegrationTest { //restart the server m_server.startServer(IntegrationUtils.prepareTestPropeties()); - MQTT mqtt = new MQTT(); - mqtt.setHost("localhost", 1883); - mqtt.setClientId("Publisher"); - m_publisher = mqtt.blockingConnection(); m_publisher.connect(); - m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), QoS.AT_MOST_ONCE, false); + m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), 0, false); } @Test public void testClientDoesntRemainSubscribedAfterASubscriptionAndServerRestart() throws Exception { - m_mqtt.setClientId("Subscriber"); //subscribe to /topic - m_subscriber = m_mqtt.blockingConnection(); m_subscriber.connect(); - Topic[] topics = new Topic[]{new Topic("/topic", QoS.AT_MOST_ONCE)}; //subscribe /topic - m_subscriber.subscribe(topics); + m_subscriber.subscribe("/topic", 0); //unsubscribe from /topic - m_subscriber.unsubscribe(new String[]{"/topic"}); + m_subscriber.unsubscribe("/topic"); m_subscriber.disconnect(); //shutdown the server @@ -150,38 +150,32 @@ public class ServerRestartIntegrationTest { //restart the server m_server.startServer(IntegrationUtils.prepareTestPropeties()); //subscriber reconnects - MQTT mqttSub = new MQTT(); - mqttSub.setHost("localhost", 1883); - mqttSub.setClientId("Subscriber"); - m_subscriber = mqttSub.blockingConnection(); + m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_dataStore); + m_subscriber.setCallback(m_messageCollector); m_subscriber.connect(); //publisher publishes on /topic - MQTT mqtt = new MQTT(); - mqtt.setHost("localhost", 1883); - mqtt.setClientId("Publisher"); - m_publisher = mqtt.blockingConnection(); + m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); m_publisher.connect(); - m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), QoS.AT_LEAST_ONCE, false); + m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), 1, false); //Expected //the subscriber doesn't get notified (it's fully unsubscribed) - assertNull(m_subscriber.receive(1, TimeUnit.SECONDS)); + assertNull(m_messageCollector.getMessage(1)); } /** * Connect subscribe to topic and publish on the same topic * */ - private BlockingConnection subscribeAndPublish(String topic) throws Exception { - BlockingConnection conn = m_mqtt.blockingConnection(); - conn.connect(); - Topic[] topics = new Topic[]{new Topic(topic, QoS.AT_MOST_ONCE)}; - conn.subscribe(topics); - conn.publish(topic, "Hello world MQTT!!".getBytes(), QoS.AT_MOST_ONCE, false); - //read the message - Message msg = conn.receive(); - msg.ack(); + private IMqttClient subscribeAndPublish(String topic) throws Exception { + IMqttClient client = new MqttClient("tcp://localhost:1883", "SubPub"); + MessageCollector collector = new MessageCollector(); + client.setCallback(collector); + client.connect(); + client.subscribe(topic, 1); + client.publish(topic, "Hello world MQTT!!".getBytes(), 0, false); + MqttMessage msg = collector.getMessage(true); assertEquals("Hello world MQTT!!", new String(msg.getPayload())); - return conn; + return client; } } \ No newline at end of file