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

- moved and cleaned many part of Paho integration tests to unit tests

Browse Source 
- fixed bug that on conflicting connect close old channel but not dropped from ConnectionDescriptors cache
This commit is contained in:
andsel
2017-06-29 17:00:14 +02:00
parent 03c4874a50
commit 23b903e2f2
7 changed files with 750 additions and 570 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java
+160 -429
View File
@@ -83,94 +83,6 @@ public class ServerIntegrationPahoTest {
m_server.stopServer();
}
@Test
public void connectWithCredentials() throws Exception {
LOG.info("*** connectWithCredentials ***");
m_client = new MqttClient("tcp://localhost:1883", "Publisher", s_pubDataStore);
MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("testuser");
options.setPassword("passwd".toCharArray());
m_client.connect(options);
assertTrue(true);
}
@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_messagesCollector.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_messagesCollector = new MessageCollector();
anotherClient.setCallback(m_messagesCollector);
anotherClient.connect(options);
anotherClient.publish("/topic", "Test my payload".getBytes(), 0, false);
assertEquals("/topic", m_messagesCollector.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_messagesCollector.getTopic());
}
/**
* Check that after a client has connected with clean session false, subscribed to some topic
* and exited, if it reconnects 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
// 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.waitMessage(1).getPayload()));
// close
m_client.disconnect();
}
@Ignore("This test hasn't any meaning using in memory storage service")
@Test
@@ -193,187 +105,6 @@ public class ServerIntegrationPahoTest {
assertEquals("/topic", m_messagesCollector.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(), 1, true);
m_client.disconnect();
// reconnect and publish
m_client.connect();
m_client.subscribe("/topic", 0);
assertEquals("/topic", m_messagesCollector.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_messagesCollector.getTopic());
m_client.unsubscribe("/topic");
m_messagesCollector.reinit();
m_client.publish("/topic", "Test my payload".getBytes(), 0, false);
assertNull(m_messagesCollector.getMessageImmediate());
}
@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_messagesCollector.getTopic());
m_client.unsubscribe("/topic");
m_client.disconnect();
m_messagesCollector.reinit();
m_client.connect(options);
m_client.publish("/topic", "Test my payload".getBytes(), 0, false);
assertNull(m_messagesCollector.getMessageImmediate());
}
@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
MqttMessage message = m_messagesCollector.waitMessage(1);
assertEquals("Hello MQTT", message.toString());
assertEquals(1, message.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_messagesCollector.waitMessage(1).toString());
}
@Test
public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws Exception {
LOG.info("*** checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession ***");
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
m_client.connect(options);
m_client.subscribe("/topic", 1);
m_client.disconnect();
m_client.connect(options);
m_client.subscribe("/topic", 1);
// publish a QoS 1 message another client publish a message on the topic
publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 1);
// Verify that after a reconnection the client receive the message
MqttMessage message = m_messagesCollector.waitMessage(1);
assertNotNull(message);
assertEquals("Hello MQTT", message.toString());
}
private void publishFromAnotherClient(String topic, byte[] payload, int qos) throws Exception {
IMqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClientPUB", s_pubDataStore);
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_messagesCollector.reinit();
m_client.connect(options);
MqttMessage message = m_messagesCollector.waitMessage(1);
assertEquals("Hello MQTT", message.toString());
assertEquals(2, message.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 2 message another client publish a message on the topic
publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 2);
m_messagesCollector.reinit();
m_client.connect(options);
assertNotNull(m_messagesCollector);
MqttMessage message = m_messagesCollector.waitMessage(1);
assertNotNull(message);
assertEquals("Hello MQTT", message.toString());
}
@Test
public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() throws Exception {
LOG.info("*** avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1, issue #16 ***");
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
m_client.connect(options);
m_client.subscribe("/topic", 1);
m_client.disconnect();
publishFromAnotherClient("/topic", "Hello MQTT 1".getBytes(), 1);
m_messagesCollector.reinit();
m_client.connect(options);
assertNotNull(m_messagesCollector);
MqttMessage message = m_messagesCollector.waitMessage(1);
assertNotNull(message);
assertEquals("Hello MQTT 1", message.toString());
m_client.disconnect();
// publish other message
publishFromAnotherClient("/topic", "Hello MQTT 2".getBytes(), 1);
// reconnect the second time
m_messagesCollector.reinit();
m_client.connect(options);
assertNotNull(m_messagesCollector);
message = m_messagesCollector.waitMessage(1);
assertNotNull(message);
assertEquals("Hello MQTT 2", message.toString());
}
/**
* subscriber A connect and subscribe on "a/b" QoS 1 subscriber B connect and subscribe on "a/+"
@@ -460,170 +191,170 @@ public class ServerIntegrationPahoTest {
assertEquals("Hello", new String(messageOnB.getPayload()));
}
@Test
public void testForceClientDisconnection_issue116() throws Exception {
LOG.info("*** testForceClientDisconnection_issue118 ***");
MessageCollector cbSubscriber1 = new MessageCollector();
MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1);
LOG.info("Connected 'subscriber' first time");
clientXA.subscribe("topic", 0);
LOG.info("Subscribed 'topic' from 'subscriber' first time");
// @Test
// public void testForceClientDisconnection_issue116() throws Exception {
// LOG.info("*** testForceClientDisconnection_issue118 ***");
// MessageCollector cbSubscriber1 = new MessageCollector();
// MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1);
// LOG.info("Connected 'subscriber' first time");
// clientXA.subscribe("topic", 0);
// LOG.info("Subscribed 'topic' from 'subscriber' first time");
//
// MqttClient clientXB = createClient("publisher", "X");
// LOG.info("Connected 'publisher' first time");
// clientXB.publish("topic", "Hello".getBytes(), 2, false);
// LOG.info("Published on 'topic' from 'publisher' first time");
//
// LOG.info("Creating second new 'subscriber'");
// MessageCollector cbSubscriber2 = new MessageCollector();
// MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2);
// LOG.info("Connected 'subscriber' second time");
// clientYA.subscribe("topic", 0);
//
// MqttClient clientYB = createClient("publisher", "Y");
// clientYB.publish("topic", "Hello 2".getBytes(), 2, true);
//
// // Verify that the second subscriber client get notified and not the first.
// assertTrue(cbSubscriber1.connectionLost());
// assertEquals("Hello 2", new String(cbSubscriber2.waitMessage(1).getPayload()));
// }
MqttClient clientXB = createClient("publisher", "X");
LOG.info("Connected 'publisher' first time");
clientXB.publish("topic", "Hello".getBytes(), 2, false);
LOG.info("Published on 'topic' from 'publisher' first time");
// @Test
// public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws Exception {
// LOG.info("*** testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142 ***");
// MqttConnectOptions options = new MqttConnectOptions();
// options.setCleanSession(false);
// m_client.connect(options);
// m_client.subscribe("/topic", 0);
// m_client.publish("/topic", "Hello".getBytes(), 0, true);
// m_client.disconnect();
//
// // second loop
// m_client.connect(options);
// m_client.subscribe("/topic", 0);
// m_client.publish("/topic", "Hello".getBytes(), 0, true);
// m_client.disconnect(); // this should give timeout
//
// assertFalse("after a disconnect the client should be disconnected", m_client.isConnected());
// }
LOG.info("Creating second new 'subscriber'");
MessageCollector cbSubscriber2 = new MessageCollector();
MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2);
LOG.info("Connected 'subscriber' second time");
clientYA.subscribe("topic", 0);
// @Test
// public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws Exception {
// LOG.info("*** checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1 ***");
// // MessageCollector m_messagesCollector = new MessageCollector();
// // m_client.setCallback(m_messagesCollector);
// m_publisher.connect();
//
// MqttConnectOptions options = new MqttConnectOptions();
// options.setCleanSession(false);
// m_client.connect(options);
// m_client.subscribe("/topic", 1);
//
// // force the publisher to send
// m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false);
//
// // read the first message and drop the connection
// MqttMessage msg = m_messagesCollector.waitMessage(1);
// assertEquals("Hello world MQTT!!-1", new String(msg.getPayload()));
// m_client.disconnect();
//
// m_publisher.publish("/topic", "Hello world MQTT!!-2".getBytes(), 1, false);
// m_publisher.publish("/topic", "Hello world MQTT!!-3".getBytes(), 1, false);
//
// // reconnect and expect to receive the hello 2 message
// m_client.connect(options);
// msg = m_messagesCollector.waitMessage(1);
// assertEquals("Hello world MQTT!!-2", new String(msg.getPayload()));
//
// msg = m_messagesCollector.waitMessage(1);
// assertEquals("Hello world MQTT!!-3", new String(msg.getPayload()));
// m_client.disconnect();
// }
MqttClient clientYB = createClient("publisher", "Y");
clientYB.publish("topic", "Hello 2".getBytes(), 2, true);
// @Test
// public void checkReplayStoredPublish_forNoCleanSession_qos1() throws Exception {
// LOG.info("*** checkReplayStoredPublish_forNoCleanSession_qos1 ***");
// MessageCollector messagesCollector = new MessageCollector();
// m_client.setCallback(messagesCollector);
// m_publisher.connect();
//
// MqttConnectOptions options = new MqttConnectOptions();
// options.setCleanSession(false);
// m_client.connect(options);
// m_client.subscribe("/topic", 1);
// m_client.disconnect();
//
// // force the publisher to send
// m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false);
//
// // reconnect and expect to receive the hello 2 message
// m_client.connect(options);
// m_client.subscribe("/topic", 1);
// MqttMessage msg = messagesCollector.waitMessage(1);
// assertNotNull(msg);
// assertEquals("Hello world MQTT!!-1", new String(msg.getPayload()));
// }
// Verify that the second subscriber client get notified and not the first.
assertTrue(cbSubscriber1.connectionLost());
assertEquals("Hello 2", new String(cbSubscriber2.waitMessage(1).getPayload()));
}
@Test
public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws Exception {
LOG.info("*** testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142 ***");
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
m_client.connect(options);
m_client.subscribe("/topic", 0);
m_client.publish("/topic", "Hello".getBytes(), 0, true);
m_client.disconnect();
// second loop
m_client.connect(options);
m_client.subscribe("/topic", 0);
m_client.publish("/topic", "Hello".getBytes(), 0, true);
m_client.disconnect(); // this should give timeout
assertFalse("after a disconnect the client should be disconnected", m_client.isConnected());
}
@Test
public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws Exception {
LOG.info("*** checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1 ***");
// MessageCollector m_messagesCollector = new MessageCollector();
// m_client.setCallback(m_messagesCollector);
m_publisher.connect();
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
m_client.connect(options);
m_client.subscribe("/topic", 1);
// force the publisher to send
m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false);
// read the first message and drop the connection
MqttMessage msg = m_messagesCollector.waitMessage(1);
assertEquals("Hello world MQTT!!-1", new String(msg.getPayload()));
m_client.disconnect();
m_publisher.publish("/topic", "Hello world MQTT!!-2".getBytes(), 1, false);
m_publisher.publish("/topic", "Hello world MQTT!!-3".getBytes(), 1, false);
// reconnect and expect to receive the hello 2 message
m_client.connect(options);
msg = m_messagesCollector.waitMessage(1);
assertEquals("Hello world MQTT!!-2", new String(msg.getPayload()));
msg = m_messagesCollector.waitMessage(1);
assertEquals("Hello world MQTT!!-3", new String(msg.getPayload()));
m_client.disconnect();
}
@Test
public void checkReplayStoredPublish_forNoCleanSession_qos1() throws Exception {
LOG.info("*** checkReplayStoredPublish_forNoCleanSession_qos1 ***");
MessageCollector messagesCollector = new MessageCollector();
m_client.setCallback(messagesCollector);
m_publisher.connect();
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
m_client.connect(options);
m_client.subscribe("/topic", 1);
m_client.disconnect();
// force the publisher to send
m_publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), 1, false);
// reconnect and expect to receive the hello 2 message
m_client.connect(options);
m_client.subscribe("/topic", 1);
MqttMessage msg = messagesCollector.waitMessage(1);
assertNotNull(msg);
assertEquals("Hello world MQTT!!-1", new String(msg.getPayload()));
}
/**
* subscriber connect and subscribe on "topic" subscriber disconnects publisher connects and
* send two message "hello1" "hello2" to "topic" subscriber connects again and receive "hello1"
* "hello2"
*/
@Test
public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() throws Exception {
LOG.info("*** checkQoS2SubscriberDisconnectReceivePersistedPublishes ***");
MessageCollector messagesCollector = new MessageCollector();
m_client.setCallback(messagesCollector);
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
m_client.connect(options);
m_client.subscribe("topic", 2);
m_client.disconnect();
m_publisher.connect();
m_publisher.publish("topic", "Hello1".getBytes(), 2, true);
m_publisher.publish("topic", "Hello2".getBytes(), 2, true);
m_publisher.disconnect();
// subscriber reconnects
m_client.connect(options);
// subscriber should receive the 2 messages missed
MqttMessage msg = messagesCollector.waitMessage(1);
assertNotNull(msg);
assertEquals("Hello1", new String(msg.getPayload()));
msg = messagesCollector.waitMessage(1);
assertNotNull(msg);
assertEquals("Hello2", new String(msg.getPayload()));
}
/**
* subscriber connect and subscribe on "a/b" QoS 1 and "a/+" QoS 2 publisher connects and send a
* message "hello" on "a/b" subscriber must receive only a single message not twice
*/
@Test
public void checkSinglePublishOnOverlappingSubscriptions() throws Exception {
LOG.info("*** checkSinglePublishOnOverlappingSubscriptions ***");
MessageCollector messagesCollector = new MessageCollector();
m_client.setCallback(messagesCollector);
m_publisher.connect();
MqttConnectOptions options = new MqttConnectOptions();
options.setCleanSession(false);
m_client.connect(options);
m_client.subscribe("a/+", 2);
m_client.subscribe("a/b", 1);
// force the publisher to send
m_publisher.publish("a/b", "Hello world MQTT!!".getBytes(), 1, false);
// reconnect and expect to receive the hello 2 message
MqttMessage msg = messagesCollector.waitMessage(1);
assertEquals("Hello world MQTT!!", new String(msg.getPayload()));
// try to listen a second publish
msg = messagesCollector.waitMessage(1);
assertNull(msg);
}
// /**
// * subscriber connect and subscribe on "topic" subscriber disconnects publisher connects and
// * send two message "hello1" "hello2" to "topic" subscriber connects again and receive "hello1"
// * "hello2"
// */
// @Test
// public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() throws Exception {
// LOG.info("*** checkQoS2SubscriberDisconnectReceivePersistedPublishes ***");
// MessageCollector messagesCollector = new MessageCollector();
// m_client.setCallback(messagesCollector);
// MqttConnectOptions options = new MqttConnectOptions();
// options.setCleanSession(false);
// m_client.connect(options);
// m_client.subscribe("topic", 2);
// m_client.disconnect();
//
// m_publisher.connect();
// m_publisher.publish("topic", "Hello1".getBytes(), 2, true);
// m_publisher.publish("topic", "Hello2".getBytes(), 2, true);
// m_publisher.disconnect();
//
// // subscriber reconnects
// m_client.connect(options);
// // subscriber should receive the 2 messages missed
// MqttMessage msg = messagesCollector.waitMessage(1);
// assertNotNull(msg);
// assertEquals("Hello1", new String(msg.getPayload()));
// msg = messagesCollector.waitMessage(1);
// assertNotNull(msg);
// assertEquals("Hello2", new String(msg.getPayload()));
// }
//
// /**
// * subscriber connect and subscribe on "a/b" QoS 1 and "a/+" QoS 2 publisher connects and send a
// * message "hello" on "a/b" subscriber must receive only a single message not twice
// */
// @Test
// public void checkSinglePublishOnOverlappingSubscriptions() throws Exception {
// LOG.info("*** checkSinglePublishOnOverlappingSubscriptions ***");
// MessageCollector messagesCollector = new MessageCollector();
// m_client.setCallback(messagesCollector);
// m_publisher.connect();
//
// MqttConnectOptions options = new MqttConnectOptions();
// options.setCleanSession(false);
// m_client.connect(options);
// m_client.subscribe("a/+", 2);
// m_client.subscribe("a/b", 1);
//
// // force the publisher to send
// m_publisher.publish("a/b", "Hello world MQTT!!".getBytes(), 1, false);
//
// // reconnect and expect to receive the hello 2 message
// MqttMessage msg = messagesCollector.waitMessage(1);
// assertEquals("Hello world MQTT!!", new String(msg.getPayload()));
//
// // try to listen a second publish
// msg = messagesCollector.waitMessage(1);
// assertNull(msg);
// }
protected MqttClient createClient(String clientName, String storeSuffix) throws MqttException {
return createClient(clientName, storeSuffix, null);
broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java
+131 -16
View File
@@ -20,21 +20,24 @@ import io.moquette.persistence.MemoryStorageService;
import io.moquette.server.netty.NettyUtils;
import io.moquette.spi.IMessagesStore;
import io.moquette.spi.ISessionsStore;
import io.moquette.spi.ISubscriptionsStore.ClientTopicCouple;
import io.moquette.spi.impl.security.PermitAllAuthorizator;
import io.moquette.spi.impl.subscriptions.Subscription;
import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory;
import io.moquette.spi.impl.subscriptions.Topic;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.mqtt.*;
import java.util.*;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.CONNECTION_ACCEPTED;
import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_LEAST_ONCE;
import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
abstract class AbstractProtocolProcessorCommonUtils {
@@ -82,20 +85,32 @@ abstract class AbstractProtocolProcessorCommonUtils {
}
protected void verifyNoPublishIsReceived() {
assertNull("No out messages from the processor", m_channel.readOutbound());
assertNull("Received an out message from processor while not expected", m_channel.readOutbound());
}
protected void subscribe(String topic, MqttQoS desiredQos) {
subscribe(this.m_channel, topic, desiredQos);
}
protected void subscribe(EmbeddedChannel channel, String topic, MqttQoS desiredQos) {
MqttSubscribeMessage subscribe = MqttMessageBuilders.subscribe()
.addSubscription(desiredQos, topic)
.messageId(1)
.build();
this.m_processor.processSubscribe(m_channel, subscribe);
MqttSubAckMessage subAck = m_channel.readOutbound();
this.m_processor.processSubscribe(channel, subscribe);
MqttSubAckMessage subAck = channel.readOutbound();
assertEquals(desiredQos.value(), (int) subAck.payload().grantedQoSLevels().get(0));
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, new Topic(topic), desiredQos);
assertTrue(subscriptions.contains(expectedSubscription));
final String clientId = NettyUtils.clientID(channel);
Subscription expectedSubscription = new Subscription(clientId, new Topic(topic), desiredQos);
verifySubscriptionExists(channel, m_sessionStore, expectedSubscription);
}
protected static void verifySubscriptionExists(Channel channel, ISessionsStore sessionsStore, Subscription expectedSubscription) {
final String clientId = NettyUtils.clientID(channel);
final Subscription subscription = sessionsStore.subscriptionStore()
.getSubscription(new ClientTopicCouple(clientId, expectedSubscription.getTopicFilter()));
assertEquals(expectedSubscription, subscription);
}
protected MqttSubAckMessage subscribeWithoutVerify(String topic, MqttQoS desiredQos) {
@@ -116,14 +131,28 @@ abstract class AbstractProtocolProcessorCommonUtils {
}
protected void unsubscribe(String topic) {
final int messageId = 1;
MqttUnsubscribeMessage msg = MqttMessageBuilders.unsubscribe()
.addTopicFilter(topic)
.messageId(1)
.messageId(messageId)
.build();
m_processor.processUnsubscribe(m_channel, msg);
}
protected void unsubscribeAndVerifyAck(String topic) {
final int messageId = 1;
MqttUnsubscribeMessage msg = MqttMessageBuilders.unsubscribe()
.addTopicFilter(topic)
.messageId(messageId)
.build();
m_processor.processUnsubscribe(m_channel, msg);
MqttUnsubAckMessage unsubAckMessageAck = m_channel.readOutbound();
assertEquals("Unsubscribe must be accepted", messageId, unsubAckMessageAck.variableHeader().messageId());
}
protected void internalPublishNotRetainedTo(String topic) {
internalPublishTo(topic, AT_MOST_ONCE, false);
}
@@ -142,7 +171,11 @@ abstract class AbstractProtocolProcessorCommonUtils {
}
protected void publishToAs(String clientId, String topic, MqttQoS qos, boolean retained) {
NettyUtils.userName(m_channel, clientId);
publishToAs(m_channel, clientId, topic, qos, retained);
}
protected void publishToAs(EmbeddedChannel channel, String clientId, String topic, MqttQoS qos, boolean retained) {
NettyUtils.userName(channel, clientId);
MqttPublishMessage publish = MqttMessageBuilders.publish()
.topicName(topic)
.retained(retained)
@@ -152,14 +185,55 @@ abstract class AbstractProtocolProcessorCommonUtils {
}
protected void publishToAs(String clientId, String topic, MqttQoS qos, int messageId, boolean retained) {
NettyUtils.userName(m_channel, clientId);
publishToAs(this.m_channel, clientId, topic, qos, messageId, retained);
}
protected void publishToAs(EmbeddedChannel channel, String clientId, String topic, MqttQoS qos, int messageId, boolean retained) {
publishToAs(channel, clientId, topic, HELLO_WORLD_MQTT, qos, messageId, retained);
}
protected void publishToAs(EmbeddedChannel channel, String clientId, String topic, String payload, MqttQoS qos, int messageId, boolean retained) {
NettyUtils.userName(channel, clientId);
MqttPublishMessage publish = MqttMessageBuilders.publish()
.topicName(topic)
.retained(retained)
.messageId(messageId)
.qos(qos)
.payload(Unpooled.copiedBuffer(HELLO_WORLD_MQTT.getBytes())).build();
this.m_processor.processPublish(m_channel, publish);
.payload(Unpooled.copiedBuffer(payload.getBytes())).build();
this.m_processor.processPublish(channel, publish);
}
protected void publishQoS2ToAs(EmbeddedChannel channel, String clientId, String topic, int messageId, boolean retained) {
publishQoS2ToAs(channel, clientId, topic, HELLO_WORLD_MQTT, messageId, retained);
}
protected void publishQoS2ToAs(EmbeddedChannel channel, String clientId, String topic, String payload, int messageId, boolean retained) {
NettyUtils.userName(channel, clientId);
MqttPublishMessage publish = MqttMessageBuilders.publish()
.topicName(topic)
.retained(retained)
.messageId(messageId)
.qos(MqttQoS.EXACTLY_ONCE)
.payload(Unpooled.copiedBuffer(payload.getBytes())).build();
this.m_processor.processPublish(channel, publish);
verifyPubrecIsReceived(channel, messageId);
MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBREL, false, AT_LEAST_ONCE, false, 0);
MqttMessage pubRel = new MqttMessage(mqttFixedHeader, from(messageId));
this.m_processor.processPubRel(channel, pubRel);
verifyPubCompIsReceived(channel, messageId);
}
private void verifyPubrecIsReceived(EmbeddedChannel channel, int messageId) {
final MqttMessage pubRec = channel.readOutbound();
assertEquals(messageId, Utils.messageId(pubRec));
}
private void verifyPubCompIsReceived(EmbeddedChannel channel, int messageId) {
final MqttMessage pubComp = channel.readOutbound();
assertEquals(messageId, Utils.messageId(pubComp));
}
@@ -168,11 +242,15 @@ abstract class AbstractProtocolProcessorCommonUtils {
}
protected void connectAsClient(String clientId) {
connectAsClient(m_channel, clientId);
}
protected void connectAsClient(EmbeddedChannel channel, String clientId) {
MqttConnectMessage connectMessage = MqttMessageBuilders.connect()
.clientId(clientId)
.build();
this.m_processor.processConnect(m_channel, connectMessage);
MqttConnAckMessage connAck = m_channel.readOutbound();
this.m_processor.processConnect(channel, connectMessage);
MqttConnAckMessage connAck = channel.readOutbound();
assertEquals("Connect must be accepted", CONNECTION_ACCEPTED, connAck.variableHeader().connectReturnCode());
}
@@ -192,8 +270,12 @@ abstract class AbstractProtocolProcessorCommonUtils {
}
protected void connectNoCleanSession() {
connectNoCleanSession(FAKE_CLIENT_ID);
}
protected void connectNoCleanSession(String clientId) {
MqttConnectMessage connectMessage = MqttMessageBuilders.connect()
.clientId(FAKE_CLIENT_ID)
.clientId(clientId)
.cleanSession(false)
.build();
this.m_processor.processConnect(m_channel, connectMessage);
@@ -201,13 +283,46 @@ abstract class AbstractProtocolProcessorCommonUtils {
assertEquals("Connect must be accepted", CONNECTION_ACCEPTED, connAck.variableHeader().connectReturnCode());
}
protected void connectWithCleanSession(String clientId) {
MqttConnectMessage connectMessage = MqttMessageBuilders.connect()
.clientId(clientId)
.cleanSession(true)
.build();
this.m_processor.processConnect(m_channel, connectMessage);
MqttConnAckMessage connAck = m_channel.readOutbound();
assertEquals("Connect must be accepted", CONNECTION_ACCEPTED, connAck.variableHeader().connectReturnCode());
}
protected void disconnect() throws InterruptedException {
this.m_processor.processDisconnect(this.m_channel);
disconnect(this.m_channel);
}
protected void disconnect(EmbeddedChannel channel) throws InterruptedException {
this.m_processor.processDisconnect(channel);
}
protected void verifyPublishIsReceived() {
final MqttPublishMessage publishReceived = m_channel.readOutbound();
verifyPublishIsReceived(m_channel);
}
protected void verifyPublishIsReceived(EmbeddedChannel channel) {
final MqttPublishMessage publishReceived = channel.readOutbound();
String payloadMessage = new String(publishReceived.payload().array());
assertEquals("Sent and received payload must be identical", HELLO_WORLD_MQTT, payloadMessage);
}
protected void verifyPublishIsReceived(MqttQoS expectedQoS) {
verifyPublishIsReceived(HELLO_WORLD_MQTT, expectedQoS);
}
protected void verifyPublishIsReceived(String expectedPayload, MqttQoS expectedQoS) {
verifyPublishIsReceived(this.m_channel, expectedPayload, expectedQoS);
}
protected void verifyPublishIsReceived(EmbeddedChannel channel, String expectedPayload, MqttQoS expectedQoS) {
final MqttPublishMessage publishReceived = channel.readOutbound();
String payloadMessage = new String(publishReceived.payload().array());
assertEquals("Sent and received payload must be identical", expectedPayload, payloadMessage);
assertEquals("Expected QoS don't match", expectedQoS, publishReceived.fixedHeader().qosLevel());
}
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+67 -110
View File
@@ -30,14 +30,12 @@ import io.netty.handler.codec.mqtt.*;
import org.junit.Before;
import org.junit.Test;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsConnAck;
import static io.moquette.spi.impl.NettyChannelAssertions.assertEqualsSubAck;
import static io.moquette.spi.impl.ProtocolProcessorTest.*;
import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*;
import static java.util.Collections.singleton;
import static java.util.Collections.singletonMap;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -63,23 +61,13 @@ public class ProtocolProcessor_CONNECT_Test {
m_sessionStore = memStorage.sessionsStore();
// m_messagesStore.initStore();
Set<String> clientIds = new HashSet<>();
clientIds.add(ProtocolProcessorTest.FAKE_CLIENT_ID);
Map<String, String> users = new HashMap<>();
users.put(ProtocolProcessorTest.TEST_USER, ProtocolProcessorTest.TEST_PWD);
m_mockAuthenticator = new MockAuthenticator(clientIds, users);
m_mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
subscriptions = new SubscriptionsDirectory();
subscriptions.init(m_sessionStore);
m_processor = new ProtocolProcessor();
m_processor.init(
subscriptions,
m_messagesStore,
m_sessionStore,
m_mockAuthenticator,
true,
new PermitAllAuthorizator(),
ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true,
new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
}
@Test
@@ -95,7 +83,7 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void testWill() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).willFlag(true)
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).willFlag(true)
.willTopic("topic").willMessage("Topic message").build();
// Exercise
@@ -114,8 +102,8 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void validAuthentication() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.TEST_USER).password(ProtocolProcessorTest.TEST_PWD).build();
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER).password(TEST_PWD).build();
// Exercise
m_processor.processConnect(m_session, msg);
@@ -127,8 +115,9 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void noPasswdAuthentication() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.TEST_USER).build();
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER)
.build();
// Exercise
m_processor.processConnect(m_session, msg);
@@ -140,8 +129,10 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void invalidAuthentication() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.TEST_USER + "_fake").password(ProtocolProcessorTest.TEST_PWD).build();
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER + "_fake")
.password(TEST_PWD)
.build();
// Exercise
m_processor.processConnect(m_session, msg);
@@ -153,15 +144,8 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void prohibitAnonymousClient() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).build();
m_processor.init(
subscriptions,
m_messagesStore,
m_sessionStore,
m_mockAuthenticator,
false,
new PermitAllAuthorizator(),
ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build();
reinitProcessorProhibitingAnonymousClients();
// Exercise
m_processor.processConnect(m_session, msg);
@@ -171,18 +155,17 @@ public class ProtocolProcessor_CONNECT_Test {
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
protected void reinitProcessorProhibitingAnonymousClients() {
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false,
new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
}
@Test
public void prohibitAnonymousClient_providingUsername() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.TEST_USER + "_fake").build();
m_processor.init(
subscriptions,
m_messagesStore,
m_sessionStore,
m_mockAuthenticator,
false,
new PermitAllAuthorizator(),
ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER + "_fake")
.build();
reinitProcessorProhibitingAnonymousClients();
// Exercise
m_processor.processConnect(m_session, msg);
@@ -194,15 +177,7 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void acceptAnonymousClient() {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).build();
m_processor.init(
subscriptions,
m_messagesStore,
m_sessionStore,
m_mockAuthenticator,
true,
new PermitAllAuthorizator(),
ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build();
// Exercise
m_processor.processConnect(m_session, msg);
@@ -214,17 +189,8 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void connectWithCleanSessionUpdateClientSession() throws InterruptedException {
m_processor.init(
subscriptions,
m_messagesStore,
m_sessionStore,
m_mockAuthenticator,
true,
new PermitAllAuthorizator(),
ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
// first connect with clean session true
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID).cleanSession(true).build();
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).cleanSession(true).build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
m_processor.processDisconnect(m_session);
@@ -232,14 +198,14 @@ public class ProtocolProcessor_CONNECT_Test {
// second connect with clean session false
m_session = new EmbeddedChannel();
MqttConnectMessage secondConnMsg = MqttMessageBuilders.connect().clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
MqttConnectMessage secondConnMsg = MqttMessageBuilders.connect().clientId(FAKE_CLIENT_ID)
.protocolVersion(MqttVersion.MQTT_3_1).build();
m_processor.processConnect(m_session, secondConnMsg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
// Verify client session is clean false
ClientSession session = m_sessionStore.sessionForClient(ProtocolProcessorTest.FAKE_CLIENT_ID);
ClientSession session = m_sessionStore.sessionForClient(FAKE_CLIENT_ID);
assertFalse(session.isCleanSession());
// Verify
@@ -249,16 +215,20 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void connectWithSameClientIDBadCredentialsDoesntDropExistingClient() {
// Connect a client1
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.TEST_USER).password(ProtocolProcessorTest.TEST_PWD).build();
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.username(TEST_USER)
.password(TEST_PWD)
.build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
// create another connect same clientID but with bad credentials
MqttConnectMessage evilClientConnMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1)
.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.EVIL_TEST_USER)
.password(ProtocolProcessorTest.EVIL_TEST_PWD).build();
MqttConnectMessage evilClientConnMsg = MqttMessageBuilders.connect()
.protocolVersion(MqttVersion.MQTT_3_1)
.clientId(FAKE_CLIENT_ID)
.username(ProtocolProcessorTest.EVIL_TEST_USER)
.password(ProtocolProcessorTest.EVIL_TEST_PWD)
.build();
EmbeddedChannel evilSession = new EmbeddedChannel();
@@ -275,9 +245,9 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void testConnAckContainsSessionPresentFlag() throws InterruptedException {
MqttConnectMessage msg = connMsg.clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
.protocolVersion(MqttVersion.MQTT_3_1_1).build();
NettyUtils.clientID(m_session, ProtocolProcessorTest.FAKE_CLIENT_ID);
NettyUtils.clientID(m_session, FAKE_CLIENT_ID);
NettyUtils.cleanSession(m_session, false);
// Connect a first time
@@ -297,7 +267,7 @@ public class ProtocolProcessor_CONNECT_Test {
@Test
public void testMultipleReconnection() throws InterruptedException {
// connect with clean a false and subscribe to a topic
MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(ProtocolProcessorTest.FAKE_CLIENT_ID)
MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(FAKE_CLIENT_ID)
.protocolVersion(MqttVersion.MQTT_3_1_1).build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(CONNECTION_ACCEPTED, m_session.readOutbound());
@@ -307,14 +277,14 @@ public class ProtocolProcessor_CONNECT_Test {
MqttSubscribeMessage subscribeMsg = MqttMessageBuilders.subscribe()
.addSubscription(MqttQoS.AT_MOST_ONCE, ProtocolProcessorTest.FAKE_TOPIC).messageId(10).build();
NettyUtils.clientID(m_session, ProtocolProcessorTest.FAKE_CLIENT_ID);
NettyUtils.clientID(m_session, FAKE_CLIENT_ID);
NettyUtils.cleanSession(m_session, false);
m_processor.processSubscribe(m_session, subscribeMsg);
Subscription expectedSubscription = new Subscription(
ProtocolProcessorTest.FAKE_CLIENT_ID,
FAKE_CLIENT_ID,
new Topic(ProtocolProcessorTest.FAKE_TOPIC),
MqttQoS.AT_MOST_ONCE);
assertTrue(subscriptions.contains(expectedSubscription));
verifySubscriptionExists(m_session, m_sessionStore, expectedSubscription);
assertEqualsSubAck(m_session.readOutbound());
// disconnect
@@ -328,22 +298,14 @@ public class ProtocolProcessor_CONNECT_Test {
assertTrue("Connection is accepted and therefore should remain open.", m_session.isOpen());
// verify that the first subscription is still preserved
assertTrue(subscriptions.contains(expectedSubscription));
verifySubscriptionExists(m_session, m_sessionStore, expectedSubscription);
}
@Test
public void testZeroByteClientIdWithoutCleanSession() {
// Allow zero byte client ids
m_processor = new ProtocolProcessor();
m_processor.init(
subscriptions,
m_messagesStore,
m_sessionStore,
m_mockAuthenticator,
true,
true,
new PermitAllAuthorizator(),
ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients();
// Connect message without clean session set to true but client id is still null
MqttConnectMessage msg = MqttMessageBuilders.connect().clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1)
@@ -358,45 +320,40 @@ public class ProtocolProcessor_CONNECT_Test {
assertFalse("Connection should be closed by the broker.", m_session.isOpen());
}
protected void reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients() {
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, true,
new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR);
}
@Test
public void testZeroByteClientIdWithCleanSession() {
// Allow zero byte client ids
m_processor = new ProtocolProcessor();
m_processor.init(
subscriptions,
m_messagesStore,
m_sessionStore,
m_mockAuthenticator,
true,
true,
new PermitAllAuthorizator(),
ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR);
reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients();
// Connect message with clean session set to true and client id is null.
MqttConnectMessage msg = connMsg.clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1).cleanSession(true)
.build();
MqttConnectMessage msg = connMsg.clientId(null)
.protocolVersion(MqttVersion.MQTT_3_1_1)
.cleanSession(true)
.build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(
"Connection should be accepted. A unique clientid should be generated"
+ " and clean session set to true.",
CONNECTION_ACCEPTED,
m_session.readOutbound());
assertEqualsConnAck("Connection MUST be accepted, unique clientid MUST be generated and clean session to true",
CONNECTION_ACCEPTED, m_session.readOutbound());
assertTrue("Connection should be valid and open,", m_session.isOpen());
}
@Test
public void testZeroByteClientIdNotAllowed() {
// Connect message with clean session set to true and client id is null.
MqttConnectMessage msg = connMsg.clientId(null).protocolVersion(MqttVersion.MQTT_3_1_1).cleanSession(true)
.build();
MqttConnectMessage msg = connMsg.clientId(null)
.protocolVersion(MqttVersion.MQTT_3_1_1)
.cleanSession(true)
.build();
m_processor.processConnect(m_session, msg);
assertEqualsConnAck(
"Zero byte client identifiers are not allowed.",
CONNECTION_REFUSED_IDENTIFIER_REJECTED,
m_session.readOutbound());
assertEqualsConnAck("Zero byte client identifiers are not allowed",
CONNECTION_REFUSED_IDENTIFIER_REJECTED, m_session.readOutbound());
assertFalse("Connection should closed.", m_session.isOpen());
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java
+377
View File
@@ -0,0 +1,377 @@
/*
* Copyright (c) 2012-2017 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.spi.impl;
import io.moquette.server.MessageCollector;
import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import static io.netty.handler.codec.mqtt.MqttQoS.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProcessorCommonUtils {
@Before
public void setUp() throws InterruptedException {
initializeProcessorAndSubsystems();
}
@Test
public void testCleanSession_maintainClientSubscriptions() throws InterruptedException {
connectAsClient(this.m_channel, "TestClient");
subscribe("/topic", AT_MOST_ONCE);
disconnect();
// reconnect and publish
EmbeddedChannel anotherChannel = new EmbeddedChannel();
// NettyUtils.clientID(anotherChannel, "TestClient");
// NettyUtils.cleanSession(anotherChannel, false);
connectAsClient(anotherChannel, "TestClient");
publishToAs(anotherChannel, "TestClient", "/topic", AT_MOST_ONCE, false);
verifyPublishIsReceived(anotherChannel);
}
@Ignore("At the end it's equal to testCleanSession_maintainClientSubscriptions")
@Test
public void testCleanSession_maintainClientSubscriptions_againstClientDestruction() throws InterruptedException {
connectAsClient("TestClient");
subscribe("/topic", AT_MOST_ONCE);
disconnect();
this.m_channel = new EmbeddedChannel();
// reconnect and publish
connectAsClient("TestClient");
publishToAs("TestClient", "/topic", AT_MOST_ONCE, false);
verifyPublishIsReceived();
}
/**
* Check that after a client has connected with clean session false, subscribed to some topic
* and exited, if it reconnects with clean session true, the broker correctly cleanup every
* previous subscription
*/
@Test
public void testCleanSession_correctlyClientSubscriptions() throws InterruptedException {
connectAsClient("TestClient");
subscribe("/topic", AT_MOST_ONCE);
disconnect();
this.m_channel = new EmbeddedChannel();
// reconnect and publish
connectWithCleanSession("TestClient");
publishToAs("TestClient", "/topic", AT_MOST_ONCE, false);
verifyNoPublishIsReceived();
}
@Test
public void testRetain_maintainMessage_againstClientDestruction() throws InterruptedException {
connect();
publishToAs(FAKE_CLIENT_ID, "/topic", AT_LEAST_ONCE, 66, true);
disconnect();
// reconnect and publish
this.m_channel = new EmbeddedChannel();
connect();
subscribe("/topic", AT_MOST_ONCE);
verifyPublishIsReceived();
}
@Test
public void testUnsubscribe_do_not_notify_anymore_same_session() {
connect();
subscribe("/topic", AT_MOST_ONCE);
publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, false);
verifyPublishIsReceived();
unsubscribeAndVerifyAck("/topic");
publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, false);
verifyNoPublishIsReceived();
}
@Test
public void testUnsubscribe_do_not_notify_anymore_new_session() throws InterruptedException {
connect();
subscribe("/topic", AT_MOST_ONCE);
publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, false);
verifyPublishIsReceived();
unsubscribeAndVerifyAck("/topic");
disconnect();
this.m_channel = new EmbeddedChannel();
connect();
publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, false);
verifyNoPublishIsReceived();
}
@Test
public void testPublishWithQoS1() throws InterruptedException {
connect();
subscribe("/topic", AT_LEAST_ONCE);
publishToAs(FAKE_CLIENT_ID, "/topic", AT_LEAST_ONCE, 66, false);
disconnect();
verifyPublishIsReceived(AT_LEAST_ONCE);
}
@Test
public void testPublishWithQoS1_notCleanSession() throws InterruptedException {
connect();
subscribe("/topic", AT_LEAST_ONCE);
disconnect();
// publish a QoS 1 message another client publish a message on the topic
publishFromAnotherClient("/topic", AT_LEAST_ONCE);
this.m_channel = new EmbeddedChannel();
connect();
verifyPublishIsReceived();
}
protected void publishFromAnotherClient(String topic, MqttQoS qos) throws InterruptedException {
EmbeddedChannel anotherChannel = new EmbeddedChannel();
connectAsClient(anotherChannel, "AnotherClient");
publishToAs(anotherChannel, "AnotherClient", topic, qos, 67, false);
disconnect(anotherChannel);
}
protected void publishFromAnotherClient(String topic, String payload, MqttQoS qos) throws InterruptedException {
EmbeddedChannel anotherChannel = new EmbeddedChannel();
connectAsClient(anotherChannel, "AnotherClient");
publishToAs(anotherChannel, "AnotherClient", topic, payload, qos, 68, false);
disconnect(anotherChannel);
}
protected void publishQoS2FromAnotherClient(String topic) throws InterruptedException {
EmbeddedChannel anotherChannel = new EmbeddedChannel();
connectAsClient(anotherChannel, "AnotherClient");
publishQoS2ToAs(anotherChannel, "AnotherClient", topic, 67, false);
disconnect(anotherChannel);
}
@Test
public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws InterruptedException {
connect();
subscribe("/topic", AT_LEAST_ONCE);
disconnect();
this.m_channel = new EmbeddedChannel();
connect();
subscribe("/topic", AT_LEAST_ONCE);
// publish a QoS 1 message another client publish a message on the topic
publishFromAnotherClient("/topic", AT_LEAST_ONCE);
// Verify that after a reconnection the client receive the message
verifyPublishIsReceived();
}
@Test
public void testPublishWithQoS2() throws InterruptedException {
connect();
subscribe("/topic", EXACTLY_ONCE);
disconnect();
publishQoS2FromAnotherClient("/topic");
this.m_channel = new EmbeddedChannel();
connect();
verifyPublishIsReceived(EXACTLY_ONCE);
}
@Test
public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() throws InterruptedException {
connect();
subscribe("/topic", AT_LEAST_ONCE);
disconnect();
publishFromAnotherClient("/topic", "Hello MQTT 1", AT_LEAST_ONCE);
this.m_channel = new EmbeddedChannel();
connect();
verifyPublishIsReceived("Hello MQTT 1", AT_LEAST_ONCE);
disconnect();
// publish other message
publishFromAnotherClient("/topic", "Hello MQTT 2", AT_LEAST_ONCE);
this.m_channel = new EmbeddedChannel();
connect();
verifyPublishIsReceived("Hello MQTT 2", AT_LEAST_ONCE);
}
@Test
public void testConnectSubPub_cycle_getTimeout_on_second_disconnect_issue142() throws InterruptedException {
connect();
subscribe("/topic", AT_MOST_ONCE);
publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, true);
disconnect();
// second loop
this.m_channel = new EmbeddedChannel();
connect();
subscribe("/topic", AT_MOST_ONCE);
publishToAs(FAKE_CLIENT_ID, "/topic", AT_MOST_ONCE, true);
disconnect();
assertFalse("after a disconnect the client should be disconnected", this.m_channel.isOpen());
}
@Test
public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws InterruptedException {
EmbeddedChannel publisher = new EmbeddedChannel();
connectAsClient(publisher, "Publisher");
connect();
subscribe("/topic", AT_LEAST_ONCE);
publishToAs(publisher, "Publisher", "/topic", "Hello world MQTT!!-1", AT_LEAST_ONCE, 99, false);
verifyPublishIsReceived("Hello world MQTT!!-1", AT_LEAST_ONCE);
disconnect();
publishToAs(publisher, "Publisher", "/topic", "Hello world MQTT!!-2", AT_LEAST_ONCE, 100, false);
publishToAs(publisher, "Publisher", "/topic", "Hello world MQTT!!-3", AT_LEAST_ONCE, 101, false);
this.m_channel = new EmbeddedChannel();
connect();
verifyPublishIsReceived("Hello world MQTT!!-2", AT_LEAST_ONCE);
verifyPublishIsReceived("Hello world MQTT!!-3", AT_LEAST_ONCE);
}
@Test
public void checkReplayStoredPublish_forNoCleanSession_qos1() throws InterruptedException {
EmbeddedChannel publisher = new EmbeddedChannel();
connectAsClient(publisher, "Publisher");
connect();
subscribe("/topic", AT_LEAST_ONCE);
disconnect();
publishToAs(publisher, "Publisher", "/topic", "Hello world MQTT!!-1", AT_LEAST_ONCE, 99, false);
this.m_channel = new EmbeddedChannel();
connect();
subscribeAndNotReadResponse("/topic", AT_LEAST_ONCE);
verifyPublishIsReceived("Hello world MQTT!!-1", AT_LEAST_ONCE);
}
/**
* subscriber connect and subscribe on "topic" subscriber disconnects publisher connects and
* send two message "hello1" "hello2" to "topic" subscriber connects again and receive "hello1"
* "hello2"
*/
@Test
public void checkQoS2SubscriberDisconnectReceivePersistedPublishes() throws InterruptedException {
connect();
subscribe("topic", EXACTLY_ONCE);
disconnect();
EmbeddedChannel publisher = new EmbeddedChannel();
connectAsClient(publisher, "Publisher");
publishQoS2ToAs(publisher, "Publisher", "topic", "Hello1", 101, true);
publishQoS2ToAs(publisher, "Publisher", "topic", "Hello2", 102, true);
disconnect(publisher);
this.m_channel = new EmbeddedChannel();
connect();
verifyPublishIsReceived("Hello1", EXACTLY_ONCE);
verifyPublishIsReceived("Hello2", EXACTLY_ONCE);
}
/**
* subscriber connect and subscribe on "a/b" QoS 1 and "a/+" QoS 2 publisher connects and send a
* message "hello" on "a/b" subscriber must receive only a single message not twice
*/
@Test
public void checkSinglePublishOnOverlappingSubscriptions() {
EmbeddedChannel publisher = new EmbeddedChannel();
connectAsClient(publisher, "Publisher");
connect();
subscribe("a/+", EXACTLY_ONCE);
subscribe("a/b", AT_LEAST_ONCE);
// force the publisher to send
publishToAs(publisher, "Publisher", "a/b", "Hello world MQTT!!", AT_LEAST_ONCE, 60, false);
verifyPublishIsReceived("Hello world MQTT!!", AT_LEAST_ONCE);
// try to listen a second publish
verifyNoPublishIsReceived();
}
@Test
public void testForceClientDisconnection_issue116() {
EmbeddedChannel clientXA = new EmbeddedChannel();
connectAsClient(clientXA, "subscriber");
subscribe(clientXA, "topic", AT_MOST_ONCE);
EmbeddedChannel clientXB = new EmbeddedChannel();
connectAsClient(clientXB, "publisher");
publishQoS2ToAs(clientXB, "publisher", "topic", "Hello", 20, false);
EmbeddedChannel clientYA = new EmbeddedChannel();
connectAsClient(clientYA, "subscriber");
subscribe(clientYA, "topic", AT_MOST_ONCE);
EmbeddedChannel clientYB = new EmbeddedChannel();
connectAsClient(clientYB, "publisher");
publishQoS2ToAs(clientYB, "publisher", "topic", "Hello 2", 20, true);
assertFalse("First 'subscriber' channel MUST be closed by the broker", clientXA.isOpen());
verifyPublishIsReceived(clientYA, "Hello 2", AT_MOST_ONCE);
// LOG.info("*** testForceClientDisconnection_issue118 ***");
// MessageCollector cbSubscriber1 = new MessageCollector();
// MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1);
// LOG.info("Connected 'subscriber' first time");
// clientXA.subscribe("topic", 0);
// LOG.info("Subscribed 'topic' from 'subscriber' first time");
//
// MqttClient clientXB = createClient("publisher", "X");
// LOG.info("Connected 'publisher' first time");
// clientXB.publish("topic", "Hello".getBytes(), 2, false);
// LOG.info("Published on 'topic' from 'publisher' first time");
//
// LOG.info("Creating second new 'subscriber'");
// MessageCollector cbSubscriber2 = new MessageCollector();
// MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2);
// LOG.info("Connected 'subscriber' second time");
// clientYA.subscribe("topic", 0);
//
// MqttClient clientYB = createClient("publisher", "Y");
// clientYB.publish("topic", "Hello 2".getBytes(), 2, true);
//
// // Verify that the second subscriber client get notified and not the first.
// assertTrue(cbSubscriber1.connectionLost());
// assertEquals("Hello 2", new String(cbSubscriber2.waitMessage(1).getPayload()));
}
}