diff --git a/benchmarking/pom.xml b/benchmarking/pom.xml index 8cfb0af2..3fd21502 100644 --- a/benchmarking/pom.xml +++ b/benchmarking/pom.xml @@ -22,7 +22,7 @@ - - - - org.slf4j - slf4j-log4j12 - 1.6.4 - - - - org.fusesource.hawtdb - hawtdb - 1.6 - - - - junit - junit - 4.8.2 - test - - - - - org.mockito - mockito-core - 1.9.0 - test - - - - org.fusesource.mqtt-client - mqtt-client - - 1.3 - test - - - - org.fusesource.hawtbuf - hawtbuf - 1.9 - - - - - - - org.apache.maven.plugins - maven-compiler-plugin - - 1.5 - 1.5 - - - - - org.apache.maven.plugins - maven-surefire-plugin - - - org/dna/mqtt/moquette/proto/TestUtils.java - org/dna/mqtt/moquette/proto/TestUtils$* - - - - - - - maven-assembly-plugin - - - - org.dna.mqtt.moquette.server.Server - - - - jar-with-dependencies - - - - - - - - - - - - - + + + 4.0.0 + + ../pom.xml + moquette-parent + org.dna.mqtt + 0.2 + + + moquette-broker + jar + 0.2 + Moquette - broker + + + UTF-8 + + + + + org.dna.mqtt + moquette-mina-parser + 0.2 + + + + org.dna.mqtt + moquette-netty-parser + 0.2 + + + + org.apache.mina + mina-core + ${mina-version} + + + + com.googlecode.disruptor + disruptor + 2.7.1 + + + + + + + org.slf4j + slf4j-log4j12 + 1.6.4 + + + + org.fusesource.hawtdb + hawtdb + 1.6 + + + + junit + junit + 4.8.2 + test + + + + + org.mockito + mockito-core + 1.9.0 + test + + + + org.fusesource.mqtt-client + mqtt-client + + 1.5 + test + + + + org.fusesource.hawtbuf + hawtbuf + 1.9 + + + + org.eclipse.paho + mqtt-client + 0.4.0 + + + + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + 1.5 + 1.5 + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + org/dna/mqtt/moquette/proto/TestUtils.java + org/dna/mqtt/moquette/proto/TestUtils$* + + + + + + + maven-assembly-plugin + + + + org.dna.mqtt.moquette.server.Server + + + + jar-with-dependencies + + + + + + + + + + + + + Paho Releases + https://repo.eclipse.org/content/repositories/paho-releases/ + + + + diff --git a/broker/src/main/resources/log4j.properties b/broker/src/main/resources/log4j.properties new file mode 100644 index 00000000..bb496264 --- /dev/null +++ b/broker/src/main/resources/log4j.properties @@ -0,0 +1,21 @@ +# Set root logger level to DEBUG and its only appender to A1. +#log4j.rootLogger=ERROR, A1 +log4j.rootLogger=ERROR, A1 +#log4j.rootLogger=WARN, A1 +#log4j.logger.org.dna.mqtt.moquette.client=WARN +#log4j.logger.org.mina=WARN +#log4j.logger.org.dna.mqtt.moquette.server=DEBUG +#log4j.logger.org.dna.mqtt.moquette.messaging=DEBUG +#log4j.logger.org.dna.mqtt.moquette.proto.MQTTLoggingFilter=DEBUG +#log4j.logger.org.dna.mqtt.moquette.proto.PublishDecoder=DEBUG + +log4j.logger.org.dna.mqtt=INFO +log4j.logger.org.dna.mqtt.moquette.server.ServerIntegrationFuseTest=DEBUG +log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=DEBUG + +# A1 is set to be a ConsoleAppender. +log4j.appender.A1=org.apache.log4j.ConsoleAppender + +# A1 uses PatternLayout. +log4j.appender.A1.layout=org.apache.log4j.PatternLayout +log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n \ No newline at end of file diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java index 11409151..62d591a9 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java @@ -5,6 +5,8 @@ import org.dna.mqtt.commons.Constants; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.io.File; import java.io.IOException; @@ -17,7 +19,9 @@ import static org.junit.Assert.*; * Test the broker using the widely used FuseSource libraries * */ public class ServerIntegrationFuseTest { - + + private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationFuseTest.class); + Server m_server; MQTT m_mqtt; BlockingConnection m_connection; @@ -34,6 +38,20 @@ public class ServerIntegrationFuseTest { m_mqtt = new MQTT(); m_mqtt.setHost("localhost", Constants.PORT); m_mqtt.setClientId("TestClient"); + m_mqtt.setTracer(new Tracer() { + public void debug(String message, Object...args) { + StringBuilder sb = new StringBuilder("["); + for (Object o : args) { + sb.append(o.toString()); + } + sb.append("]"); + if (args.length > 0) { + LOG.debug(message + sb); + } else { + LOG.debug(message); + } + } + }); } @After @@ -51,7 +69,7 @@ public class ServerIntegrationFuseTest { @Test public void testSubscribe() throws Exception { - System.out.println("*** testSubscribe ***"); + LOG.info("*** testSubscribe ***"); m_connection = m_mqtt.blockingConnection(); m_connection.connect(); @@ -67,7 +85,7 @@ public class ServerIntegrationFuseTest { @Test public void testCleanSession_maintainClientSubscriptions() throws Exception { - System.out.println("*** testCleanSession_maintainClientSubscriptions ***"); + LOG.info("*** testCleanSession_maintainClientSubscriptions ***"); MQTT localMqtt = new MQTT(m_mqtt); localMqtt.setCleanSession(false); BlockingConnection connection = localMqtt.blockingConnection(); @@ -89,7 +107,7 @@ public class ServerIntegrationFuseTest { @Test public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() throws Exception { - System.out.println("*** testCleanSession_maintainClientSubscriptions_againstClientDestruction ***"); + LOG.info("*** testCleanSession_maintainClientSubscriptions_againstClientDestruction ***"); MQTT localMqtt = new MQTT(m_mqtt); localMqtt.setCleanSession(false); BlockingConnection connection = localMqtt.blockingConnection(); @@ -117,7 +135,7 @@ public class ServerIntegrationFuseTest { */ @Test public void testCleanSession_correctlyClientSubscriptions() throws Exception { - System.out.println("*** testCleanSession_correctlyClientSubscriptions ***"); + LOG.info("*** testCleanSession_correctlyClientSubscriptions ***"); MQTT localMqtt = new MQTT(m_mqtt); localMqtt.setCleanSession(false); BlockingConnection connection = localMqtt.blockingConnection(); @@ -138,7 +156,7 @@ public class ServerIntegrationFuseTest { @Test public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws Exception { - System.out.println("*** testCleanSession_maintainClientSubscriptions_withServerRestart ***"); + LOG.info("*** testCleanSession_maintainClientSubscriptions_withServerRestart ***"); MQTT localMqtt = new MQTT(m_mqtt); localMqtt.setCleanSession(false); BlockingConnection connection = localMqtt.blockingConnection(); @@ -160,12 +178,13 @@ public class ServerIntegrationFuseTest { //Verify that the message is published due to prevoius subscription Message msg = anotherconnection.receive(1, TimeUnit.SECONDS); + msg.ack(); assertEquals("/topic", msg.getTopic()); } @Test public void testRetain_maintainMessage_againstClientDestruction() throws Exception { - System.out.println("*** testRetain_maintainMessage_againstClientDestruction ***"); + LOG.info("*** testRetain_maintainMessage_againstClientDestruction ***"); m_connection = m_mqtt.blockingConnection(); m_connection.connect(); m_connection.publish("/topic", "Test my payload".getBytes(), QoS.AT_MOST_ONCE, true); @@ -182,7 +201,7 @@ public class ServerIntegrationFuseTest { @Test public void testUnsubscribe_do_not_notify_anymore_same_session() throws Exception { - System.out.println("*** testUnsubscribe_do_not_notify_anymore_same_session ***"); + LOG.info("*** testUnsubscribe_do_not_notify_anymore_same_session ***"); m_connection = m_mqtt.blockingConnection(); m_connection.connect(); m_connection.subscribe(new Topic[] {new Topic("/topic", QoS.AT_MOST_ONCE)}); @@ -202,7 +221,7 @@ public class ServerIntegrationFuseTest { @Test public void testUnsubscribe_do_not_notify_anymore_new_session() throws Exception { - System.out.println("*** testUnsubscribe_do_not_notify_anymore_new_session ***"); + LOG.info("*** testUnsubscribe_do_not_notify_anymore_new_session ***"); m_connection = m_mqtt.blockingConnection(); m_connection.connect(); m_connection.subscribe(new Topic[] {new Topic("/topic", QoS.AT_MOST_ONCE)}); @@ -225,7 +244,7 @@ public class ServerIntegrationFuseTest { @Test public void testPublishWithQoS1() throws Exception { - System.out.println("*** testPublishWithQoS1 ***"); + LOG.info("*** testPublishWithQoS1 ***"); m_connection = m_mqtt.blockingConnection(); m_connection.connect(); @@ -245,7 +264,7 @@ public class ServerIntegrationFuseTest { @Test public void testPublishWithQoS1_notCleanSession() throws Exception { - System.out.println("*** testPublishWithQoS1_notCleanSession ***"); + LOG.info("*** testPublishWithQoS1_notCleanSession ***"); MQTT mqtt = new MQTT(m_mqtt); mqtt.setCleanSession(false); m_connection = mqtt.blockingConnection(); @@ -285,7 +304,7 @@ public class ServerIntegrationFuseTest { @Test public void testPublishWithQoS2() throws Exception { - System.out.println("*** testPublishWithQoS2 ***"); + LOG.info("*** testPublishWithQoS2 ***"); m_mqtt.setClientId("TestClient"); m_connection = m_mqtt.blockingConnection(); m_connection.connect(); @@ -306,7 +325,7 @@ public class ServerIntegrationFuseTest { @Test public void testPublishReceiveWithQoS2() throws Exception { - System.out.println("*** testPublishReceiveWithQoS2 ***"); + LOG.info("*** testPublishReceiveWithQoS2 ***"); m_mqtt.setCleanSession(false); m_connection = m_mqtt.blockingConnection(); m_connection.connect(); 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 new file mode 100644 index 00000000..f0e2b8aa --- /dev/null +++ b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java @@ -0,0 +1,323 @@ +package org.dna.mqtt.moquette.server; + +import org.eclipse.paho.client.mqttv3.*; +import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; +import org.junit.*; +import org.slf4j.Logger; +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() { + try { + m_latch.await(1, TimeUnit.SECONDS); + } 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); + + static MqttClientPersistence s_dataStore; + + Server m_server; + IMqttClient m_client; + TestCallback m_callback; + + @BeforeClass + public static void beforeTests() { + String tmpDir = System.getProperty("java.io.tmpdir"); + s_dataStore = new MqttDefaultFilePersistence(tmpDir); + } + + protected void startServer() throws IOException { + m_server = new Server(); + m_server.startServer(); + } + + @Before + public void setUp() throws Exception { + startServer(); + + m_client = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); + m_callback = new TestCallback(); + m_client.setCallback(m_callback); + } + + @After + public void tearDown() throws Exception { + if (m_client.isConnected()) { + m_client.disconnect(); + } + + m_server.stopServer(); + File dbFile = new File(Server.STORAGE_FILE_PATH); + if (dbFile.exists()) { + dbFile.delete(); + } + } + + @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_callback.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_callback = new TestCallback(); + anotherClient.setCallback(m_callback); + anotherClient.connect(options); + anotherClient.publish("/topic", "Test my payload".getBytes(), 0, false); + + assertEquals("/topic", m_callback.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_callback.getTopic()); + } + + /** + * Check that after a client has connected with clean session false, subscribed + * to some topic and exited, if it reconnect 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 + m_client.connect(); + m_client.publish("/topic", "Test my payload".getBytes(), 0, false); + + assertNull(m_callback.getMessage()); + } + + @Test + public void testCleanSession_maintainClientSubscriptions_withServerRestart() throws Exception { + LOG.info("*** testCleanSession_maintainClientSubscriptions_withServerRestart ***"); + MqttConnectOptions options = new MqttConnectOptions(); + options.setCleanSession(false); + m_client.connect(options); + m_client.subscribe("/topic", 0); + m_client.disconnect(); + + m_server.stopServer(); + + m_server.startServer(); + + //reconnect and publish + m_client.connect(options); + m_client.publish("/topic", "Test my payload".getBytes(), 0, false); + + assertEquals("/topic", m_callback.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(), 0, true); + + m_client.disconnect(); + + //reconnect and publish + m_client.connect(); + m_client.subscribe("/topic", 0); + + assertEquals("/topic", m_callback.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_callback.getTopic()); + + m_client.unsubscribe("/topic"); + m_callback.reinit(); + m_client.publish("/topic", "Test my payload".getBytes(), 0, false); + + assertNull(m_callback.getMessage()); + } + + @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_callback.getTopic()); + + m_client.unsubscribe("/topic"); + m_client.disconnect(); + + m_callback.reinit(); + m_client.connect(options); + m_client.publish("/topic", "Test my payload".getBytes(), 0, false); + + assertNull(m_callback.getMessage()); + } + + @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 + assertEquals("Hello MQTT", m_callback.getMessage().toString()); + assertEquals(1, m_callback.getMessage().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_callback.getMessage().toString()); + } + + private void publishFromAnotherClient(String topic, byte[] payload, int qos) throws Exception { + IMqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClientPUB", s_dataStore); + 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_client.connect(options); + + assertEquals("Hello MQTT", m_callback.getMessage().toString()); + assertEquals(2, m_callback.getMessage().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 1 message another client publish a message on the topic + publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 2); + + m_client.connect(options); + + assertEquals("Hello MQTT", m_callback.getMessage().toString()); + } + + +} + diff --git a/parser/pom.xml b/parser/pom.xml index 7609c266..36c14c6b 100644 --- a/parser/pom.xml +++ b/parser/pom.xml @@ -1,130 +1,130 @@ - - - 4.0.0 - - ../pom.xml - moquette-parent - org.dna.mqtt - 0.2 - - - moquette-mina-parser - jar - 0.2 - Moquette - MINA Parser - - - UTF-8 - - - - - org.apache.mina - mina-core - 2.0.4 - - - - org.dna.mqtt - moquette-parser-commons - 0.2 - - - - - - - - - org.slf4j - slf4j-log4j12 - 1.6.4 - - - - - - junit - junit - 4.8.2 - test - - - - - org.mockito - mockito-core - 1.9.0 - test - - - - - - - - - org.apache.maven.plugins - maven-compiler-plugin - - 1.5 - 1.5 - - - - - org.apache.maven.plugins - maven-surefire-plugin - - - org/dna/mqtt/moquette/proto/TestUtils.java - org/dna/mqtt/moquette/proto/TestUtils$* - - - - - - - - - - - - + + + 4.0.0 + + ../pom.xml + moquette-parent + org.dna.mqtt + 0.2 + + + moquette-mina-parser + jar + 0.2 + Moquette - MINA Parser + + + UTF-8 + + + + + org.apache.mina + mina-core + ${mina-version} + + + + org.dna.mqtt + moquette-parser-commons + 0.2 + + + + + + + + + org.slf4j + slf4j-log4j12 + 1.6.4 + + + + + + junit + junit + 4.8.2 + test + + + + + org.mockito + mockito-core + 1.9.0 + test + + + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + + 1.5 + 1.5 + + + + + org.apache.maven.plugins + maven-surefire-plugin + + + org/dna/mqtt/moquette/proto/TestUtils.java + org/dna/mqtt/moquette/proto/TestUtils$* + + + + + + + + + + + + diff --git a/pom.xml b/pom.xml index 0c2343a3..9c16b3db 100644 --- a/pom.xml +++ b/pom.xml @@ -1,26 +1,27 @@ - - - 4.0.0 - - 0.2 - - - org.dna.mqtt - moquette-parent - - pom - 0.2 - Moquette MQTT library - http://code.google.com/p/moquette-mqtt/ - - - - parser_commons - parser - netty_parser - broker - benchmarking - - - + + + 4.0.0 + + 0.2 + 2.0.7 + + + org.dna.mqtt + moquette-parent + + pom + 0.2 + Moquette MQTT library + http://code.google.com/p/moquette-mqtt/ + + + + parser_commons + parser + netty_parser + broker + benchmarking + + +