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

Moved test to use a test client callback that queue up all received messages

Browse Source
This commit is contained in:
andsel
2016-03-06 19:50:47 +01:00
parent 6cadaebb0c
commit 4e8f72a34b
8 changed files with 71 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/pom.xml
-20
View File
@@ -44,26 +44,6 @@
<scope>runtime</scope>
</dependency>
<!--dependency>
<groupId>org.apache.felix</groupId>
<artifactId>org.osgi.core</artifactId>
<version>1.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.felix</groupId>
<artifactId>org.osgi.compendium</artifactId>
<version>1.0.0</version>
</dependency-->
<!-- Included to embedded an HawtDB dependency, we host commons-logging -->
<!--dependency>
<groupId>commons-logging</groupId>
<artifactId>commons-logging</artifactId>
<version>1.1.3</version>
</dependency-->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-common</artifactId>
broker/src/test/java/io/moquette/server/MessageCollector.java
+9 -5
View File
@@ -43,11 +43,7 @@ public class MessageCollector implements MqttCallback {
if (!checkElapsed && m_messages.isEmpty()) {
return null;
}
try {
return m_messages.poll(1, TimeUnit.SECONDS).message;
} catch (InterruptedException e) {
return null;
}
return getMessage(1);
}
public MqttMessage getMessage(int delay) {
@@ -89,7 +85,15 @@ public class MessageCollector implements MqttCallback {
m_messages.offer(new ReceivedMessage(message, topic));
}
/**
* Invoked when the message sent to a server is ACKED (PUBACK or PUBCOMP by the server)
* */
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
// try {
// m_messages.offer(new ReceivedMessage(token.getMessage(), token.getTopics()[0]));
// } catch (MqttException e) {
// e.printStackTrace();
// }
}
}
broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java
+5 -4
View File
@@ -44,7 +44,7 @@ public class ServerIntegrationEmbeddedPublishTest {
Server m_server;
IMqttClient m_subscriber;
TestCallback m_callback;
MessageCollector m_callback;
IConfig m_config;
protected void startServer() throws IOException {
@@ -69,7 +69,7 @@ public class ServerIntegrationEmbeddedPublishTest {
startServer();
m_subscriber = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore);
m_callback = new TestCallback();
m_callback = new MessageCollector();
m_subscriber.setCallback(m_callback);
}
@@ -269,7 +269,7 @@ public class ServerIntegrationEmbeddedPublishTest {
verifyMessageIsReceivedSuccessfully();
}
@Test(expected = IllegalStateException.class)
@Test
public void testClientSubscribeAfterDisconnected() throws Exception {
LOG.info("*** testClientSubscribeAfterDisconnected ***");
@@ -280,6 +280,7 @@ public class ServerIntegrationEmbeddedPublishTest {
//verifyMessageIsReceivedSuccessfully();
//m_subscriber.receive(2, TimeUnit.MILLISECONDS);
m_callback.getMessage(true);
MqttMessage message = m_callback.getMessage(true);
assertTrue(message == null);
}
}
broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java
+50 -43
View File
@@ -42,7 +42,7 @@ public class ServerIntegrationPahoTest {
Server m_server;
IMqttClient m_client;
IMqttClient m_publisher;
TestCallback m_callback;
MessageCollector m_messagesCollector;
IConfig m_config;
@BeforeClass
@@ -67,8 +67,8 @@ public class ServerIntegrationPahoTest {
startServer();
m_client = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore);
m_callback = new TestCallback();
m_client.setCallback(m_callback);
m_messagesCollector = new MessageCollector();
m_client.setCallback(m_messagesCollector);
m_publisher = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore);
}
@@ -114,7 +114,7 @@ public class ServerIntegrationPahoTest {
message.setRetained(false);
m_client.publish("/topic", message);
assertEquals("/topic", m_callback.getTopic());
assertEquals("/topic", m_messagesCollector.getTopic());
}
@@ -129,12 +129,12 @@ public class ServerIntegrationPahoTest {
//reconnect and publish
MqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore);
m_callback = new TestCallback();
anotherClient.setCallback(m_callback);
m_messagesCollector = new MessageCollector();
anotherClient.setCallback(m_messagesCollector);
anotherClient.connect(options);
anotherClient.publish("/topic", "Test my payload".getBytes(), 0, false);
assertEquals("/topic", m_callback.getTopic());
assertEquals("/topic", m_messagesCollector.getTopic());
}
@@ -151,7 +151,7 @@ public class ServerIntegrationPahoTest {
m_client.connect(options);
m_client.publish("/topic", "Test my payload".getBytes(), 0, false);
assertEquals("/topic", m_callback.getTopic());
assertEquals("/topic", m_messagesCollector.getTopic());
}
/**
@@ -169,10 +169,17 @@ public class ServerIntegrationPahoTest {
m_client.disconnect();
//reconnect and publish
//options.setCleanSession(false);
m_client = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore);
MessageCollector subListener = new MessageCollector();
m_client.setCallback(subListener);
m_client.connect(options);
m_client.publish("/topic", "Test my payload".getBytes(), 0, false);
//Ugly, we need to get self notified after the publish
assertEquals("Test my payload", new String(subListener.getMessage(1).getPayload()));
assertEquals("Test my payload", new String(m_callback.getMessage(false).getPayload()));
//close
m_client.disconnect();
}
@Test
@@ -192,7 +199,7 @@ public class ServerIntegrationPahoTest {
m_client.connect(options);
m_client.publish("/topic", "Test my payload".getBytes(), 0, false);
assertEquals("/topic", m_callback.getTopic());
assertEquals("/topic", m_messagesCollector.getTopic());
}
@Test
@@ -206,7 +213,7 @@ public class ServerIntegrationPahoTest {
m_client.connect();
m_client.subscribe("/topic", 0);
assertEquals("/topic", m_callback.getTopic());
assertEquals("/topic", m_messagesCollector.getTopic());
}
@Test
@@ -218,13 +225,13 @@ public class ServerIntegrationPahoTest {
m_client.subscribe("/topic", 0);
m_client.publish("/topic", "Test my payload".getBytes(), 0, false);
//m_client.disconnect();
assertEquals("/topic", m_callback.getTopic());
assertEquals("/topic", m_messagesCollector.getTopic());
m_client.unsubscribe("/topic");
m_callback.reinit();
m_messagesCollector.reinit();
m_client.publish("/topic", "Test my payload".getBytes(), 0, false);
assertNull(m_callback.getMessage(false));
assertNull(m_messagesCollector.getMessage(false));
}
@Test
@@ -236,16 +243,16 @@ public class ServerIntegrationPahoTest {
m_client.subscribe("/topic", 0);
m_client.publish("/topic", "Test my payload".getBytes(), 0, false);
//m_client.disconnect();
assertEquals("/topic", m_callback.getTopic());
assertEquals("/topic", m_messagesCollector.getTopic());
m_client.unsubscribe("/topic");
m_client.disconnect();
m_callback.reinit();
m_messagesCollector.reinit();
m_client.connect(options);
m_client.publish("/topic", "Test my payload".getBytes(), 0, false);
assertNull(m_callback.getMessage(false));
assertNull(m_messagesCollector.getMessage(false));
}
@Test
@@ -257,7 +264,7 @@ public class ServerIntegrationPahoTest {
m_client.disconnect();
//reconnect and publish
MqttMessage message = m_callback.getMessage(true);
MqttMessage message = m_messagesCollector.getMessage(true);
assertEquals("Hello MQTT", message.toString());
assertEquals(1, message.getQos());
}
@@ -276,7 +283,7 @@ public class ServerIntegrationPahoTest {
m_client.connect(options);
assertEquals("Hello MQTT", m_callback.getMessage(true).toString());
assertEquals("Hello MQTT", m_messagesCollector.getMessage(true).toString());
}
@Test
@@ -295,7 +302,7 @@ public class ServerIntegrationPahoTest {
publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 1);
//Verify that after a reconnection the client receive the message
MqttMessage message = m_callback.getMessage(true);
MqttMessage message = m_messagesCollector.getMessage(true);
assertNotNull(message);
assertEquals("Hello MQTT", message.toString());
}
@@ -318,10 +325,10 @@ public class ServerIntegrationPahoTest {
//publish a QoS 1 message another client publish a message on the topic
publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 2);
m_callback.reinit();
m_messagesCollector.reinit();
m_client.connect(options);
MqttMessage message = m_callback.getMessage(true);
MqttMessage message = m_messagesCollector.getMessage(true);
assertEquals("Hello MQTT", message.toString());
assertEquals(2, message.getQos());
}
@@ -337,11 +344,11 @@ public class ServerIntegrationPahoTest {
//publish a QoS 2 message another client publish a message on the topic
publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 2);
m_callback.reinit();
m_messagesCollector.reinit();
m_client.connect(options);
assertNotNull(m_callback);
MqttMessage message = m_callback.getMessage(true);
assertNotNull(m_messagesCollector);
MqttMessage message = m_messagesCollector.getMessage(true);
assertNotNull(message);
assertEquals("Hello MQTT", message.toString());
}
@@ -356,11 +363,11 @@ public class ServerIntegrationPahoTest {
m_client.disconnect();
publishFromAnotherClient("/topic", "Hello MQTT 1".getBytes(), 1);
m_callback.reinit();
m_messagesCollector.reinit();
m_client.connect(options);
assertNotNull(m_callback);
MqttMessage message = m_callback.getMessage(true);
assertNotNull(m_messagesCollector);
MqttMessage message = m_messagesCollector.getMessage(true);
assertNotNull(message);
assertEquals("Hello MQTT 1", message.toString());
m_client.disconnect();
@@ -369,10 +376,10 @@ public class ServerIntegrationPahoTest {
publishFromAnotherClient("/topic", "Hello MQTT 2".getBytes(), 1);
//reconnect the second time
m_callback.reinit();
m_messagesCollector.reinit();
m_client.connect(options);
assertNotNull(m_callback);
message = m_callback.getMessage(true);
assertNotNull(m_messagesCollector);
message = m_messagesCollector.getMessage(true);
assertNotNull(message);
assertEquals("Hello MQTT 2", message.toString());
}
@@ -392,7 +399,7 @@ public class ServerIntegrationPahoTest {
MqttClientPersistence dsSubscriberA = new MqttDefaultFilePersistence(tmpDir + File.separator + "subscriberA");
MqttClient subscriberA = new MqttClient("tcp://localhost:1883", "SubscriberA", dsSubscriberA);
TestCallback cbSubscriberA = new TestCallback();
MessageCollector cbSubscriberA = new MessageCollector();
subscriberA.setCallback(cbSubscriberA);
subscriberA.connect();
subscriberA.subscribe("a/b", 1);
@@ -400,7 +407,7 @@ public class ServerIntegrationPahoTest {
MqttClientPersistence dsSubscriberB = new MqttDefaultFilePersistence(tmpDir + File.separator + "subscriberB");
MqttClient subscriberB = new MqttClient("tcp://localhost:1883", "SubscriberB", dsSubscriberB);
TestCallback cbSubscriberB = new TestCallback();
MessageCollector cbSubscriberB = new MessageCollector();
subscriberB.setCallback(cbSubscriberB);
subscriberB.connect();
subscriberB.subscribe("a/+", 2);
@@ -428,7 +435,7 @@ public class ServerIntegrationPahoTest {
MqttClientPersistence dsSubscriberA = new MqttDefaultFilePersistence(tmpDir + File.separator + "clientForSubscribe1");
MqttClient clientForSubscribe1 = new MqttClient("tcp://localhost:1883", "clientForSubscribe1", dsSubscriberA);
TestCallback cbSubscriber1 = new TestCallback();
MessageCollector cbSubscriber1 = new MessageCollector();
clientForSubscribe1.setCallback(cbSubscriber1);
clientForSubscribe1.connect();
clientForSubscribe1.subscribe("topic", 0);
@@ -442,7 +449,7 @@ public class ServerIntegrationPahoTest {
//clientForSubscribe2 connect and subscribe to /topic QoS2
MqttClientPersistence dsSubscriberB = new MqttDefaultFilePersistence(tmpDir + File.separator + "clientForSubscribe2");
MqttClient clientForSubscribe2 = new MqttClient("tcp://localhost:1883", "clientForSubscribe2", dsSubscriberB);
TestCallback cbSubscriber2 = new TestCallback();
MessageCollector cbSubscriber2 = new MessageCollector();
clientForSubscribe2.setCallback(cbSubscriber2);
clientForSubscribe2.connect();
clientForSubscribe2.subscribe("topic", 0);
@@ -465,14 +472,14 @@ public class ServerIntegrationPahoTest {
@Test
public void testForceClientDisconnection_issue116() throws Exception {
LOG.info("*** testForceClientDisconnection_issue118 ***");
TestCallback cbSubscriber1 = new TestCallback();
MessageCollector cbSubscriber1 = new MessageCollector();
MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1);
clientXA.subscribe("topic", 0);
MqttClient clientXB = createClient("publisher", "X");
clientXB.publish("topic", "Hello".getBytes(), 2, false);
TestCallback cbSubscriber2 = new TestCallback();
MessageCollector cbSubscriber2 = new MessageCollector();
MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2);
clientYA.subscribe("topic", 0);
@@ -506,8 +513,8 @@ public class ServerIntegrationPahoTest {
@Test
public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws Exception {
LOG.info("*** checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1 ***");
MessageCollector messagesCollector = new MessageCollector();
m_client.setCallback(messagesCollector);
// MessageCollector m_messagesCollector = new MessageCollector();
// m_client.setCallback(m_messagesCollector);
m_publisher.connect();
MqttConnectOptions options = new MqttConnectOptions();
@@ -519,7 +526,7 @@ public class ServerIntegrationPahoTest {
m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false);
//read the first message and drop the connection
MqttMessage msg = messagesCollector.getMessage(true);
MqttMessage msg = m_messagesCollector.getMessage(true);
assertEquals("Hello world MQTT!!-1", new String(msg.getPayload()));
m_client.disconnect();
@@ -528,10 +535,10 @@ public class ServerIntegrationPahoTest {
//reconnect and expect to receive the hello 2 message
m_client.connect(options);
msg = messagesCollector.getMessage(true);
msg = m_messagesCollector.getMessage(true);
assertEquals("Hello world MQTT!!-2", new String(msg.getPayload()));
msg = messagesCollector.getMessage(true);
msg = m_messagesCollector.getMessage(true);
assertEquals("Hello world MQTT!!-3", new String(msg.getPayload()));
m_client.disconnect();
}
@@ -628,7 +635,7 @@ public class ServerIntegrationPahoTest {
return createClient(clientName, storeSuffix, null);
}
protected MqttClient createClient(String clientName, String storeSuffix, TestCallback cb) throws MqttException {
protected MqttClient createClient(String clientName, String storeSuffix, MessageCollector cb) throws MqttException {
String tmpDir = System.getProperty("java.io.tmpdir");
//clientX connect and subscribe to /topic QoS2
MqttClientPersistence dsClient = new MqttDefaultFilePersistence(tmpDir + File.separator + storeSuffix + clientName);
broker/src/test/java/io/moquette/server/ServerIntegrationQoSValidationTest.java
+2 -2
View File
@@ -50,7 +50,7 @@ public class ServerIntegrationQoSValidationTest {
IMqttClient m_subscriber;
IMqttClient m_publisher;
TestCallback m_callback;
MessageCollector m_callback;
IConfig m_config;
@BeforeClass
@@ -72,7 +72,7 @@ public class ServerIntegrationQoSValidationTest {
startServer();
m_subscriber = new MqttClient("tcp://localhost:1883", "Subscriber", s_subDataStore);
m_callback = new TestCallback();
m_callback = new MessageCollector();
m_subscriber.setCallback(m_callback);
m_subscriber.connect();
broker/src/test/java/io/moquette/server/ServerIntegrationSSLTest.java
+2 -2
View File
@@ -61,7 +61,7 @@ public class ServerIntegrationSSLTest {
static MqttClientPersistence s_dataStore;
IMqttClient m_client;
TestCallback m_callback;
MessageCollector m_callback;
@BeforeClass
public static void beforeTests() {
@@ -94,7 +94,7 @@ public class ServerIntegrationSSLTest {
m_client = new MqttClient("ssl://localhost:8883", "TestClient", s_dataStore);
// m_client = new MqttClient("ssl://test.mosquitto.org:8883", "TestClient", s_dataStore);
m_callback = new TestCallback();
m_callback = new MessageCollector();
m_client.setCallback(m_callback);
}
broker/src/test/java/io/moquette/server/TestCallback.java
-82
View File
@@ -1,82 +0,0 @@
/*
* Copyright (c) 2012-2015 The original author or authors
* ------------------------------------------------------
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
* and Apache License v2.0 which accompanies this distribution.
*
* The Eclipse Public License is available at
* http://www.eclipse.org/legal/epl-v10.html
*
* The Apache License v2.0 is available at
* http://www.opensource.org/licenses/apache2.0.php
*
* You may elect to redistribute this code under either of these licenses.
*/
package io.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);
private boolean m_connectionLost = false;
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();
}
return m_message;
}
public String getTopic() {
try {
m_latch.await(5, TimeUnit.SECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
return m_topic;
}
void reinit() {
m_latch = new CountDownLatch(1);
m_message = null;
m_topic = null;
m_connectionLost = false;
}
public boolean connectionLost() {
return m_connectionLost;
}
@Override
public void connectionLost(Throwable cause) {
m_connectionLost = true;
}
@Override
public void messageArrived(String topic, MqttMessage message) throws Exception {
m_message = message;
m_topic = topic;
m_latch.countDown();
}
@Override
public void deliveryComplete(IMqttDeliveryToken token) {
}
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+2
View File
@@ -197,11 +197,13 @@ public class ProtocolProcessorTest {
//Verify
firstReceiverChannel.flush();
PublishMessage pub2FirstSubscriber = (PublishMessage) firstReceiverChannel.readOutbound();
assertNotNull(pub2FirstSubscriber);
String firstMessageContent = DebugUtils.payload2Str(pub2FirstSubscriber.getPayload());
assertEquals("Hello", firstMessageContent);
secondReceiverChannel.flush();
PublishMessage pub2SecondSubscriber = (PublishMessage) secondReceiverChannel.readOutbound();
assertNotNull(pub2SecondSubscriber);
String secondMessageContent = DebugUtils.payload2Str(pub2SecondSubscriber.getPayload());