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

Minor, rewritten test to user Paho instead of fusesource mqtt client

Browse Source
This commit is contained in:
andsel
2016-03-06 19:50:47 +01:00
parent efbfab7914
commit 6cadaebb0c
3 changed files with 71 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/test/java/io/moquette/server/MessageCollector.java
+1 -4
View File
@@ -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");
}
}
}
broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java
+14 -13
View File
@@ -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;
broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java
+56 -62
View File
@@ -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;
}
}