anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

* Reimplemented the Fuse integration tests into Paho client lib

Browse Source 
* Extracted common pom properties
This commit is contained in:
selva.andre
2013-09-19 20:14:07 +02:00
parent f09fcd6cd5
commit 81b6516cac
7 changed files with 701 additions and 323 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java
+32 -13
View File
@@ -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();
broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java
+323
View File
@@ -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());
}
}