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

Added integration test to check firing of will message (issue #171)

Browse Source
This commit is contained in:
andsel
2016-05-02 20:48:18 +02:00
parent 5029da8763
commit dbe9c7bcaa
1 changed files with 49 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/test/java/io/moquette/server/ServerLowlevelMessagesIntegrationTests.java
+48 -21
View File
@@ -22,10 +22,7 @@ import io.moquette.parser.proto.messages.ConnAckMessage;
import io.moquette.server.config.IConfig; import io.moquette.server.config.IConfig;
import io.moquette.server.config.MemoryConfig; import io.moquette.server.config.MemoryConfig;
import io.moquette.testclient.Client; import io.moquette.testclient.Client;
import org.eclipse.paho.client.mqttv3.IMqttClient; import org.eclipse.paho.client.mqttv3.*;
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.After;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
@@ -51,6 +48,7 @@ public class ServerLowlevelMessagesIntegrationTests {
IMqttClient m_willSubscriber; IMqttClient m_willSubscriber;
MessageCollector m_messageCollector; MessageCollector m_messageCollector;
IConfig m_config; IConfig m_config;
AbstractMessage receivedMsg;
protected void startServer() throws IOException { protected void startServer() throws IOException {
m_server = new Server(); m_server = new Server();
@@ -102,27 +100,13 @@ public class ServerLowlevelMessagesIntegrationTests {
m_willSubscriber.connect(); m_willSubscriber.connect();
m_willSubscriber.subscribe(willTestamentTopic, 0); m_willSubscriber.subscribe(willTestamentTopic, 0);
int keepAlive = 2; //secs connect(willTestamentTopic, willTestamentMsg);
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProtocolVersion((byte) 3);
connectMessage.setClientID("FAKECLNT");
connectMessage.setKeepAlive(keepAlive);
connectMessage.setWillFlag(true);
connectMessage.setWillMessage(willTestamentMsg.getBytes());
connectMessage.setWillTopic(willTestamentTopic);
connectMessage.setWillQos(QOSType.MOST_ONE.byteValue());
//Execute
m_client.sendMessage(connectMessage);
long connectTime = System.currentTimeMillis(); long connectTime = System.currentTimeMillis();
//but after the 2 KEEP ALIVE timeout expires it gets fired, //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 //NB it's 1,5 * KEEP_ALIVE so 3 secs and some millis to propagate the message
MqttMessage msg = m_messageCollector.getMessage(3300); MqttMessage msg = m_messageCollector.getMessage(3300);
long willMessageReceiveTime = System.currentTimeMillis(); long willMessageReceiveTime = System.currentTimeMillis();
if (msg == null) {
LOG.warn("testament message is null");
}
assertNotNull("the will message should be fired after keep alive!", msg); assertNotNull("the will message should be fired after keep alive!", msg);
//the will message hasn't to be received before the elapsing of Keep Alive timeout //the will message hasn't to be received before the elapsing of Keep Alive timeout
assertTrue(willMessageReceiveTime - connectTime > 3000); assertTrue(willMessageReceiveTime - connectTime > 3000);
@@ -131,8 +115,6 @@ public class ServerLowlevelMessagesIntegrationTests {
m_willSubscriber.disconnect(); m_willSubscriber.disconnect();
} }
AbstractMessage receivedMsg;
@Test @Test
public void checkRejectConnectWithEmptyClientID() throws InterruptedException { public void checkRejectConnectWithEmptyClientID() throws InterruptedException {
LOG.info("*** checkRejectConnectWithEmptyClientID ***"); LOG.info("*** checkRejectConnectWithEmptyClientID ***");
@@ -163,4 +145,49 @@ public class ServerLowlevelMessagesIntegrationTests {
assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, connAck.getReturnCode()); assertEquals(ConnAckMessage.IDENTIFIER_REJECTED, connAck.getReturnCode());
} }
@Test
public void testWillMessageIsPublishedOnClientBadDisconnection() throws InterruptedException, MqttException {
LOG.info("*** testWillMessageIsPublishedOnClientBadDisconnection ***");
String willTestamentTopic = "/will/test";
String willTestamentMsg = "Bye bye";
m_willSubscriber.connect();
m_willSubscriber.subscribe(willTestamentTopic, 0);
connect(willTestamentTopic, willTestamentMsg);
//kill will publisher
m_client.close();
//Verify will testament is published
MqttMessage receivedTestament = m_messageCollector.getMessage(1000);
assertEquals(willTestamentMsg, new String(receivedTestament.getPayload()));
m_willSubscriber.disconnect();
}
private void connect(String willTestamentTopic, String willTestamentMsg) throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(1);
m_client.setCallback(new Client.ICallback() {
public void call(AbstractMessage msg) {
receivedMsg = msg;
latch.countDown();
}
});
int keepAlive = 2; //secs
ConnectMessage connectMessage = new ConnectMessage();
connectMessage.setProtocolVersion((byte) 3);
connectMessage.setClientID("FAKECLNT");
connectMessage.setKeepAlive(keepAlive);
connectMessage.setWillFlag(true);
connectMessage.setWillMessage(willTestamentMsg.getBytes());
connectMessage.setWillTopic(willTestamentTopic);
connectMessage.setWillQos(QOSType.MOST_ONE.byteValue());
m_client.sendMessage(connectMessage);
latch.await(200, TimeUnit.MILLISECONDS);
assertTrue("conn-ack message is received", this.receivedMsg instanceof ConnAckMessage);
}
} }