diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java index 9cef50e9..d3520cf8 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java @@ -8,59 +8,10 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; import static org.junit.Assert.*; public class ServerIntegrationPahoTest { - static class TestCallback implements MqttCallback { - - private MqttMessage m_message; - private String m_topic; - private CountDownLatch m_latch = new CountDownLatch(1); - - public MqttMessage getMessage(boolean checkElapsed) { - try { - boolean elapsed = !m_latch.await(1, TimeUnit.SECONDS); - if (elapsed && checkElapsed) { - throw new IllegalStateException("Elapsed the timeout to get the result"); - } - } catch (InterruptedException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. - } - return m_message; - } - - public String getTopic() { - try { - m_latch.await(5, TimeUnit.SECONDS); - } catch (InterruptedException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. - } - return m_topic; - } - - void reinit() { - m_latch = new CountDownLatch(1); - m_message = null; - m_topic = null; - } - - public void connectionLost(Throwable cause) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void messageArrived(String topic, MqttMessage message) throws Exception { - m_message = message; - m_topic = topic; - m_latch.countDown(); - } - - public void deliveryComplete(IMqttDeliveryToken token) { - //To change body of implemented methods use File | Settings | File Templates. - } - } private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationSuite.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationSuite.java deleted file mode 100644 index 7c12667a..00000000 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationSuite.java +++ /dev/null @@ -1,231 +0,0 @@ -package org.dna.mqtt.moquette.server; - - -import java.io.File; -import java.io.IOException; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import org.eclipse.paho.client.mqttv3.IMqttClient; -import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; -import org.eclipse.paho.client.mqttv3.MqttCallback; -import org.eclipse.paho.client.mqttv3.MqttClient; -import org.eclipse.paho.client.mqttv3.MqttClientPersistence; -import org.eclipse.paho.client.mqttv3.MqttMessage; -import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; -import org.junit.After; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNull; -import org.junit.Before; -import org.junit.Test; -import org.junit.BeforeClass; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * - * @author andrea - */ -public class ServerIntegrationQoSValidationSuite { - static class TestCallback implements MqttCallback { - - private MqttMessage m_message; - private String m_topic; - private CountDownLatch m_latch = new CountDownLatch(1); - - public MqttMessage getMessage(boolean checkElapsed) { - try { - boolean elapsed = !m_latch.await(1, TimeUnit.SECONDS); - if (elapsed && checkElapsed) { - throw new IllegalStateException("Elapsed the timeout to get the result"); - } - } catch (InterruptedException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. - } - return m_message; - } - - public String getTopic() { - try { - m_latch.await(5, TimeUnit.SECONDS); - } catch (InterruptedException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. - } - return m_topic; - } - - void reinit() { - m_latch = new CountDownLatch(1); - m_message = null; - m_topic = null; - } - - public void connectionLost(Throwable cause) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void messageArrived(String topic, MqttMessage message) throws Exception { - m_message = message; - m_topic = topic; - m_latch.countDown(); - } - - public void deliveryComplete(IMqttDeliveryToken token) { - //To change body of implemented methods use File | Settings | File Templates. - } - } - - - private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class); - - Server m_server; - static MqttClientPersistence s_subDataStore; - static MqttClientPersistence s_pubDataStore; - - IMqttClient m_subscriber; - IMqttClient m_publisher; - TestCallback m_callback; - - @BeforeClass - public static void beforeTests() { - String tmpDir = System.getProperty("java.io.tmpdir"); - s_subDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "subscriber"); - s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher"); - } - - protected void startServer() throws IOException { - m_server = new Server(); - m_server.startServer(); - } - - @Before - public void setUp() throws Exception { - startServer(); - - m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_subDataStore); - m_callback = new TestCallback(); - m_subscriber.setCallback(m_callback); - m_subscriber.connect(); - - m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore); -// m_callback = new TestCallback(); -// m_subscriber.setCallback(m_callback); - m_publisher.connect(); - } - - @After - public void tearDown() throws Exception { - if (m_publisher.isConnected()) { - m_publisher.disconnect(); - } - - if (m_subscriber.isConnected()) { - m_subscriber.disconnect(); - } - - m_server.stopServer(); - File dbFile = new File(Server.STORAGE_FILE_PATH); - if (dbFile.exists()) { - dbFile.delete(); - } - } - - @Test - public void checkSubscriberQoS0ReceiveQoS0publishes() throws Exception { - LOG.info("*** checkSubscriberQoS0ReceiveQoS0publishes ***"); - m_subscriber.subscribe("/topic", 0); - - m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(), 0, false); - MqttMessage message = m_callback.getMessage(true); - assertEquals("Hello world MQTT QoS0", message.toString()); - assertEquals(0, message.getQos()); - } - - @Test - public void checkSubscriberQoS0ReceiveQoS1publishes_downgrade() throws Exception { - LOG.info("*** checkSubscriberQoS0ReceiveQoS1publishes_downgrade ***"); - m_subscriber.subscribe("/topic", 0); - - m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(), 1, false); - MqttMessage message = m_callback.getMessage(true); - assertEquals("Hello world MQTT QoS1", message.toString()); - assertEquals(0, message.getQos()); - } - - @Test - public void checkSubscriberQoS0ReceiveQoS2publishes_downgrade() throws Exception { - LOG.info("*** checkSubscriberQoS0ReceiveQoS2publishes_downgrade ***"); - m_subscriber.subscribe("/topic", 0); - - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); - MqttMessage message = m_callback.getMessage(true); - assertEquals("Hello world MQTT QoS2", message.toString()); - assertEquals(0, message.getQos()); - } - - @Test - public void checkSubscriberQoS1ReceiveQoS0publishes() throws Exception { - LOG.info("*** checkSubscriberQoS1ReceiveQoS0publishes ***"); - m_subscriber.subscribe("/topic", 1); - - m_publisher.publish("/topic", "Hello world MQTT QoS0".getBytes(), 0, false); - MqttMessage message = m_callback.getMessage(true); - assertEquals("Hello world MQTT QoS0", message.toString()); - assertEquals(0, message.getQos()); - } - - @Test - public void checkSubscriberQoS1ReceiveQoS1publishes() throws Exception { - LOG.info("*** checkSubscriberQoS1ReceiveQoS1publishes ***"); - m_subscriber.subscribe("/topic", 1); - - m_publisher.publish("/topic", "Hello world MQTT QoS1".getBytes(), 1, false); - MqttMessage message = m_callback.getMessage(true); - assertEquals("Hello world MQTT QoS1", message.toString()); - assertEquals(1, message.getQos()); - } - - @Test - public void checkSubscriberQoS1ReceiveQoS2publishes_downgrade() throws Exception { - LOG.info("*** checkSubscriberQoS1ReceiveQoS2publishes_downgrade ***"); - m_subscriber.subscribe("/topic", 1); - - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); - MqttMessage message = m_callback.getMessage(true); - assertEquals("Hello world MQTT QoS2", message.toString()); - assertEquals(1, message.getQos()); - } - - @Test - public void checkSubscriberQoS2ReceiveQoS0publishes() throws Exception { - LOG.info("*** checkSubscriberQoS2ReceiveQoS0publishes ***"); - m_subscriber.subscribe("/topic", 2); - - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 0, false); - MqttMessage message = m_callback.getMessage(true); - assertEquals("Hello world MQTT QoS2", message.toString()); - assertEquals(0, message.getQos()); - } - - @Test - public void checkSubscriberQoS2ReceiveQoS1publishes() throws Exception { - LOG.info("*** checkSubscriberQoS2ReceiveQoS1publishes ***"); - m_subscriber.subscribe("/topic", 2); - - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 1, false); - MqttMessage message = m_callback.getMessage(true); - assertEquals("Hello world MQTT QoS2", message.toString()); - assertEquals(1, message.getQos()); - } - - @Test - public void checkSubscriberQoS2ReceiveQoS2publishes() throws Exception { - LOG.info("*** checkSubscriberQoS2ReceiveQoS2publishes ***"); - m_subscriber.subscribe("/topic", 2); - - m_publisher.publish("/topic", "Hello world MQTT QoS2".getBytes(), 2, false); - MqttMessage message = m_callback.getMessage(true); - assertEquals("Hello world MQTT QoS2", message.toString()); - assertEquals(2, message.getQos()); - } - -} diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationTest.java index a6d90b8b..2645717d 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationQoSValidationTest.java @@ -3,11 +3,7 @@ package org.dna.mqtt.moquette.server; import java.io.File; import java.io.IOException; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; import org.eclipse.paho.client.mqttv3.IMqttClient; -import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; -import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttClient; import org.eclipse.paho.client.mqttv3.MqttClientPersistence; import org.eclipse.paho.client.mqttv3.MqttMessage; @@ -25,54 +21,6 @@ import org.slf4j.LoggerFactory; * @author andrea */ public class ServerIntegrationQoSValidationTest { - static class TestCallback implements MqttCallback { - - private MqttMessage m_message; - private String m_topic; - private CountDownLatch m_latch = new CountDownLatch(1); - - public MqttMessage getMessage(boolean checkElapsed) { - try { - boolean elapsed = !m_latch.await(1, TimeUnit.SECONDS); - if (elapsed && checkElapsed) { - throw new IllegalStateException("Elapsed the timeout to get the result"); - } - } catch (InterruptedException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. - } - return m_message; - } - - public String getTopic() { - try { - m_latch.await(5, TimeUnit.SECONDS); - } catch (InterruptedException e) { - e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. - } - return m_topic; - } - - void reinit() { - m_latch = new CountDownLatch(1); - m_message = null; - m_topic = null; - } - - public void connectionLost(Throwable cause) { - //To change body of implemented methods use File | Settings | File Templates. - } - - public void messageArrived(String topic, MqttMessage message) throws Exception { - m_message = message; - m_topic = topic; - m_latch.countDown(); - } - - public void deliveryComplete(IMqttDeliveryToken token) { - //To change body of implemented methods use File | Settings | File Templates. - } - } - private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class); diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/TestCallback.java b/broker/src/test/java/org/dna/mqtt/moquette/server/TestCallback.java new file mode 100644 index 00000000..bcc352d2 --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/TestCallback.java @@ -0,0 +1,59 @@ +package org.dna.mqtt.moquette.server; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; +import org.eclipse.paho.client.mqttv3.MqttCallback; +import org.eclipse.paho.client.mqttv3.MqttMessage; + +/** + * + * @author andrea + */ +class TestCallback implements MqttCallback { + + private MqttMessage m_message; + private String m_topic; + private CountDownLatch m_latch = new CountDownLatch(1); + + public MqttMessage getMessage(boolean checkElapsed) { + try { + boolean elapsed = !m_latch.await(1, TimeUnit.SECONDS); + if (elapsed && checkElapsed) { + throw new IllegalStateException("Elapsed the timeout to get the result"); + } + } catch (InterruptedException e) { + e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + } + return m_message; + } + + public String getTopic() { + try { + m_latch.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + e.printStackTrace(); //To change body of catch statement use File | Settings | File Templates. + } + return m_topic; + } + + void reinit() { + m_latch = new CountDownLatch(1); + m_message = null; + m_topic = null; + } + + public void connectionLost(Throwable cause) { + //To change body of implemented methods use File | Settings | File Templates. + } + + public void messageArrived(String topic, MqttMessage message) throws Exception { + m_message = message; + m_topic = topic; + m_latch.countDown(); + } + + public void deliveryComplete(IMqttDeliveryToken token) { + //To change body of implemented methods use File | Settings | File Templates. + } + }