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

broker/ProtocolProcessorTest: code formatter used

Browse Source
This commit is contained in:
Andreas Mager
2017-02-01 17:13:43 +01:00
parent 5125132f23
commit a0a758949e
1 changed files with 170 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+170 -184
View File
@@ -13,6 +13,7 @@
* *
* You may elect to redistribute this code under either of these licenses. * You may elect to redistribute this code under either of these licenses.
*/ */
package io.moquette.spi.impl; package io.moquette.spi.impl;
import io.moquette.interception.InterceptHandler; import io.moquette.interception.InterceptHandler;
@@ -30,10 +31,8 @@ import io.netty.channel.embedded.EmbeddedChannel;
import io.netty.handler.codec.mqtt.*; import io.netty.handler.codec.mqtt.*;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import java.util.*; import java.util.*;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
import static io.moquette.spi.impl.NettyChannelAssertions.assertConnAckAccepted; import static io.moquette.spi.impl.NettyChannelAssertions.assertConnAckAccepted;
import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired; import static io.moquette.spi.impl.ProtocolProcessor.lowerQosToTheSubscriptionDesired;
import static org.junit.Assert.*; import static org.junit.Assert.*;
@@ -44,19 +43,20 @@ import static org.mockito.Mockito.*;
* @author andrea * @author andrea
*/ */
public class ProtocolProcessorTest { public class ProtocolProcessorTest {
final static String FAKE_CLIENT_ID = "FAKE_123";
final static String FAKE_CLIENT_ID2 = "FAKE_456";
final static String FAKE_PUBLISHER_ID = "Publisher";
final static String FAKE_TOPIC = "/news";
final static String BAD_FORMATTED_TOPIC = "#MQTTClient";
final static String TEST_USER = "fakeuser"; static final String FAKE_CLIENT_ID = "FAKE_123";
final static String TEST_PWD = "fakepwd"; static final String FAKE_CLIENT_ID2 = "FAKE_456";
final static String EVIL_TEST_USER = "eviluser"; static final String FAKE_PUBLISHER_ID = "Publisher";
final static String EVIL_TEST_PWD = "unsecret"; static final String FAKE_TOPIC = "/news";
static final String BAD_FORMATTED_TOPIC = "#MQTTClient";
final static List<InterceptHandler> EMPTY_OBSERVERS = Collections.emptyList(); static final String TEST_USER = "fakeuser";
final static BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); static final String TEST_PWD = "fakepwd";
static final String EVIL_TEST_USER = "eviluser";
static final String EVIL_TEST_PWD = "unsecret";
static final List<InterceptHandler> EMPTY_OBSERVERS = Collections.emptyList();
static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS);
EmbeddedChannel m_channel; EmbeddedChannel m_channel;
MqttConnectMessage connMsg; MqttConnectMessage connMsg;
@@ -69,26 +69,24 @@ public class ProtocolProcessorTest {
@Before @Before
public void setUp() throws InterruptedException { public void setUp() throws InterruptedException {
/*connMsg = new ConnectMessage(); /*
connMsg.setProtocolVersion((byte) 0x03); * connMsg = new ConnectMessage(); connMsg.setProtocolVersion((byte) 0x03);
*
connMsg = MessageBuilder.connect() * connMsg = MessageBuilder.connect() .protocolVersion(MqttVersion.MQTT_3_1)
.protocolVersion(MqttVersion.MQTT_3_1) * .clientId(FAKE_CLIENT_ID) .cleanSession(true) .build();
.clientId(FAKE_CLIENT_ID) */
.cleanSession(true)
.build();*/
m_channel = new EmbeddedChannel(); m_channel = new EmbeddedChannel();
NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); NettyUtils.clientID(m_channel, FAKE_CLIENT_ID);
NettyUtils.cleanSession(m_channel, false); NettyUtils.cleanSession(m_channel, false);
//sleep to let the messaging batch processor to process the initEvent // sleep to let the messaging batch processor to process the initEvent
Thread.sleep(300); Thread.sleep(300);
MemoryStorageService memStorage = new MemoryStorageService(); MemoryStorageService memStorage = new MemoryStorageService();
memStorage.initStore(); memStorage.initStore();
m_messagesStore = memStorage.messagesStore(); m_messagesStore = memStorage.messagesStore();
m_sessionStore = memStorage.sessionsStore(); m_sessionStore = memStorage.sessionsStore();
//m_messagesStore.initStore(); // m_messagesStore.initStore();
Set<String> clientIds = new HashSet<>(); Set<String> clientIds = new HashSet<>();
clientIds.add(FAKE_CLIENT_ID); clientIds.add(FAKE_CLIENT_ID);
@@ -100,18 +98,23 @@ public class ProtocolProcessorTest {
subscriptions = new SubscriptionsStore(); subscriptions = new SubscriptionsStore();
subscriptions.init(memStorage.sessionsStore()); subscriptions.init(memStorage.sessionsStore());
m_processor = new ProtocolProcessor(); m_processor = new ProtocolProcessor();
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, m_processor.init(
new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); subscriptions,
m_messagesStore,
m_sessionStore,
m_mockAuthenticator,
true,
new PermitAllAuthorizator(),
NO_OBSERVERS_INTERCEPTOR);
} }
@Test @Test
public void testPublishToItself() throws InterruptedException { public void testPublishToItself() throws InterruptedException {
final Subscription subscription = new Subscription(FAKE_CLIENT_ID, final Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
//subscriptions.matches(topic) redefine the method to return true // subscriptions.matches(topic) redefine the method to return true
SubscriptionsStore subs = new SubscriptionsStore() { SubscriptionsStore subs = new SubscriptionsStore() {
@Override @Override
public List<Subscription> matches(String topic) { public List<Subscription> matches(String topic) {
if (topic.equals(FAKE_TOPIC)) { if (topic.equals(FAKE_TOPIC)) {
@@ -121,45 +124,43 @@ public class ProtocolProcessorTest {
} }
} }
}; };
//simulate a connect that register a clientID to an IoSession // simulate a connect that register a clientID to an IoSession
MemoryStorageService storageService = new MemoryStorageService(); MemoryStorageService storageService = new MemoryStorageService();
storageService.initStore(); storageService.initStore();
subs.init(storageService.sessionsStore()); subs.init(storageService.sessionsStore());
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); m_processor.init(
MqttConnectMessage connectMessage = MessageBuilder.connect() subs,
.protocolVersion(MqttVersion.MQTT_3_1) m_messagesStore,
.clientId(FAKE_CLIENT_ID) m_sessionStore,
.cleanSession(true) null,
.build(); true,
new PermitAllAuthorizator(),
NO_OBSERVERS_INTERCEPTOR);
MqttConnectMessage connectMessage = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1)
.clientId(FAKE_CLIENT_ID).cleanSession(true).build();
m_processor.processConnect(m_channel, connectMessage); m_processor.processConnect(m_channel, connectMessage);
// Exercise
//Exercise MqttPublishMessage msg = MessageBuilder.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
MqttPublishMessage msg = MessageBuilder.publish() .retained(false).payload("Hello".getBytes()).build();
.topicName(FAKE_TOPIC)
.qos(MqttQoS.AT_MOST_ONCE)
.retained(false)
.payload("Hello".getBytes())
.build();
NettyUtils.userName(m_channel, "FakeCLI"); NettyUtils.userName(m_channel, "FakeCLI");
m_processor.processPublish(m_channel, msg); m_processor.processPublish(m_channel, msg);
//Verify // Verify
assertNotNull(m_channel.readOutbound()); assertNotNull(m_channel.readOutbound());
//TODO check received message attributes // TODO check received message attributes
} }
@Test @Test
public void testPublishToMultipleSubscribers() throws InterruptedException { public void testPublishToMultipleSubscribers() throws InterruptedException {
final Subscription subscription = new Subscription(FAKE_CLIENT_ID, final Subscription subscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
FAKE_TOPIC, MqttQoS.AT_MOST_ONCE); final Subscription subscriptionClient2 = new Subscription(FAKE_CLIENT_ID2, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
final Subscription subscriptionClient2 = new Subscription(FAKE_CLIENT_ID2,
FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
//subscriptions.matches(topic) redefine the method to return true // subscriptions.matches(topic) redefine the method to return true
SubscriptionsStore subs = new SubscriptionsStore() { SubscriptionsStore subs = new SubscriptionsStore() {
@Override @Override
public List<Subscription> matches(String topic) { public List<Subscription> matches(String topic) {
if (topic.equals(FAKE_TOPIC)) { if (topic.equals(FAKE_TOPIC)) {
@@ -169,44 +170,40 @@ public class ProtocolProcessorTest {
} }
} }
}; };
//simulate a connect that register a clientID to an IoSession // simulate a connect that register a clientID to an IoSession
MemoryStorageService storageService = new MemoryStorageService(); MemoryStorageService storageService = new MemoryStorageService();
storageService.initStore(); storageService.initStore();
subs.init(storageService.sessionsStore()); subs.init(storageService.sessionsStore());
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); m_processor.init(
subs,
m_messagesStore,
m_sessionStore,
null,
true,
new PermitAllAuthorizator(),
NO_OBSERVERS_INTERCEPTOR);
EmbeddedChannel firstReceiverChannel = new EmbeddedChannel(); EmbeddedChannel firstReceiverChannel = new EmbeddedChannel();
MqttConnectMessage connectMessage = MessageBuilder.connect() MqttConnectMessage connectMessage = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1)
.protocolVersion(MqttVersion.MQTT_3_1) .clientId(FAKE_CLIENT_ID).cleanSession(true).build();
.clientId(FAKE_CLIENT_ID)
.cleanSession(true)
.build();
m_processor.processConnect(firstReceiverChannel, connectMessage); m_processor.processConnect(firstReceiverChannel, connectMessage);
assertConnAckAccepted(firstReceiverChannel); assertConnAckAccepted(firstReceiverChannel);
//connect the second fake subscriber // connect the second fake subscriber
EmbeddedChannel secondReceiverChannel = new EmbeddedChannel(); EmbeddedChannel secondReceiverChannel = new EmbeddedChannel();
MqttConnectMessage connectMessage2 = MessageBuilder.connect() MqttConnectMessage connectMessage2 = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1)
.protocolVersion(MqttVersion.MQTT_3_1) .clientId(FAKE_CLIENT_ID2).cleanSession(true).build();
.clientId(FAKE_CLIENT_ID2)
.cleanSession(true)
.build();
m_processor.processConnect(secondReceiverChannel, connectMessage2); m_processor.processConnect(secondReceiverChannel, connectMessage2);
assertConnAckAccepted(secondReceiverChannel); assertConnAckAccepted(secondReceiverChannel);
//Exercise // Exercise
MqttPublishMessage msg = MessageBuilder.publish() MqttPublishMessage msg = MessageBuilder.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.topicName(FAKE_TOPIC) .retained(false).payload("Hello".getBytes()).build();
.qos(MqttQoS.AT_MOST_ONCE)
.retained(false)
.payload("Hello".getBytes())
.build();
NettyUtils.userName(m_channel, "FakeCLI"); NettyUtils.userName(m_channel, "FakeCLI");
m_processor.processPublish(m_channel, msg); m_processor.processPublish(m_channel, msg);
// Verify
//Verify
firstReceiverChannel.flush(); firstReceiverChannel.flush();
MqttPublishMessage pub2FirstSubscriber = firstReceiverChannel.readOutbound(); MqttPublishMessage pub2FirstSubscriber = firstReceiverChannel.readOutbound();
assertNotNull(pub2FirstSubscriber); assertNotNull(pub2FirstSubscriber);
@@ -222,15 +219,13 @@ public class ProtocolProcessorTest {
@Test @Test
public void testSubscribe() { public void testSubscribe() {
MqttSubscribeMessage msg = MessageBuilder.subscribe() MqttSubscribeMessage msg = MessageBuilder.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC)
.addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC) .messageId(10).build();
.messageId(10) // Exercise
.build();
//Exercise
m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false);
m_processor.processSubscribe(m_channel, msg); m_processor.processSubscribe(m_channel, msg);
//Verify // Verify
assertTrue(m_channel.readOutbound() instanceof MqttSubAckMessage); assertTrue(m_channel.readOutbound() instanceof MqttSubAckMessage);
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE); Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
assertTrue(subscriptions.contains(expectedSubscription)); assertTrue(subscriptions.contains(expectedSubscription));
@@ -243,19 +238,23 @@ public class ProtocolProcessorTest {
IAuthorizator mockAuthorizator = mock(IAuthorizator.class); IAuthorizator mockAuthorizator = mock(IAuthorizator.class);
when(mockAuthorizator.canRead(eq(FAKE_TOPIC), eq(fakeUserName), eq(FAKE_CLIENT_ID))).thenReturn(false); when(mockAuthorizator.canRead(eq(FAKE_TOPIC), eq(fakeUserName), eq(FAKE_CLIENT_ID))).thenReturn(false);
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, m_processor.init(
mockAuthorizator, NO_OBSERVERS_INTERCEPTOR); subscriptions,
m_messagesStore,
m_sessionStore,
m_mockAuthenticator,
true,
mockAuthorizator,
NO_OBSERVERS_INTERCEPTOR);
//Exercise // Exercise
MqttSubscribeMessage msg = MessageBuilder.subscribe() MqttSubscribeMessage msg = MessageBuilder.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC)
.addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC) .messageId(10).build();
.messageId(10)
.build();
m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false);
m_processor.processSubscribe(m_channel, msg); m_processor.processSubscribe(m_channel, msg);
//Verify // Verify
Object ackMsg = m_channel.readOutbound(); Object ackMsg = m_channel.readOutbound();
assertTrue(ackMsg instanceof MqttSubAckMessage); assertTrue(ackMsg instanceof MqttSubAckMessage);
MqttSubAckMessage subAckMsg = (MqttSubAckMessage) ackMsg; MqttSubAckMessage subAckMsg = (MqttSubAckMessage) ackMsg;
@@ -270,39 +269,34 @@ public class ProtocolProcessorTest {
@Test @Test
public void testDoubleSubscribe() { public void testDoubleSubscribe() {
MqttSubscribeMessage msg = MessageBuilder.subscribe() MqttSubscribeMessage msg = MessageBuilder.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC)
.addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC) .messageId(10).build();
.messageId(10)
.build();
m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false);
assertEquals(0, subscriptions.size()); assertEquals(0, subscriptions.size());
m_processor.processSubscribe(m_channel, msg);
//Exercise
m_processor.processSubscribe(m_channel, msg); m_processor.processSubscribe(m_channel, msg);
//Verify // Exercise
m_processor.processSubscribe(m_channel, msg);
// Verify
assertEquals(1, subscriptions.size()); assertEquals(1, subscriptions.size());
Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE); Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
assertTrue(subscriptions.contains(expectedSubscription)); assertTrue(subscriptions.contains(expectedSubscription));
} }
@Test @Test
public void testSubscribeWithBadFormattedTopic() { public void testSubscribeWithBadFormattedTopic() {
MqttSubscribeMessage msg = MessageBuilder.subscribe() MqttSubscribeMessage msg = MessageBuilder.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, BAD_FORMATTED_TOPIC)
.addSubscription(MqttQoS.AT_MOST_ONCE, BAD_FORMATTED_TOPIC) .messageId(10).build();
.messageId(10)
.build();
m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false);
assertEquals(0, subscriptions.size()); assertEquals(0, subscriptions.size());
//Exercise // Exercise
m_processor.processSubscribe(m_channel, msg); m_processor.processSubscribe(m_channel, msg);
//Verify // Verify
assertEquals(0, subscriptions.size()); assertEquals(0, subscriptions.size());
Object recvSubAckMessage = m_channel.readOutbound(); Object recvSubAckMessage = m_channel.readOutbound();
assertTrue(recvSubAckMessage instanceof MqttSubAckMessage); assertTrue(recvSubAckMessage instanceof MqttSubAckMessage);
@@ -311,30 +305,27 @@ public class ProtocolProcessorTest {
/* /*
* Check topicFilter is a valid MQTT topic filter (issue 68) * Check topicFilter is a valid MQTT topic filter (issue 68)
* */ */
@Test @Test
public void testUnsubscribeWithBadFormattedTopic() { public void testUnsubscribeWithBadFormattedTopic() {
MqttUnsubscribeMessage msg = MessageBuilder.unsubscribe() MqttUnsubscribeMessage msg = MessageBuilder.unsubscribe().addTopicFilter(BAD_FORMATTED_TOPIC).messageId(1)
.addTopicFilter(BAD_FORMATTED_TOPIC)
.messageId(1)
.build(); .build();
//Exercise // Exercise
m_processor.processUnsubscribe(m_channel, msg); m_processor.processUnsubscribe(m_channel, msg);
//Verify // Verify
assertFalse("If client unsubscribe with bad topic than channel must be closed", m_channel.isOpen()); assertFalse("If client unsubscribe with bad topic than channel must be closed", m_channel.isOpen());
} }
@Test @Test
public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception { public void testPublishOfRetainedMessage_afterNewSubscription() throws Exception {
//simulate a connect that register a clientID to an IoSession // simulate a connect that register a clientID to an IoSession
final Subscription subscription = new Subscription(FAKE_PUBLISHER_ID, final Subscription subscription = new Subscription(FAKE_PUBLISHER_ID, FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
FAKE_TOPIC, MqttQoS.AT_MOST_ONCE);
//subscriptions.matches(topic) redefine the method to return true // subscriptions.matches(topic) redefine the method to return true
SubscriptionsStore subs = new SubscriptionsStore() { SubscriptionsStore subs = new SubscriptionsStore() {
@Override @Override
public List<Subscription> matches(String topic) { public List<Subscription> matches(String topic) {
if (topic.equals(FAKE_TOPIC)) { if (topic.equals(FAKE_TOPIC)) {
@@ -348,36 +339,34 @@ public class ProtocolProcessorTest {
storageService.initStore(); storageService.initStore();
subs.init(storageService.sessionsStore()); subs.init(storageService.sessionsStore());
//simulate a connect that register a clientID to an IoSession // simulate a connect that register a clientID to an IoSession
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); m_processor.init(
MqttConnectMessage connectMessage = MessageBuilder.connect() subs,
.clientId(FAKE_PUBLISHER_ID) m_messagesStore,
.protocolVersion(MqttVersion.MQTT_3_1) m_sessionStore,
.cleanSession(true) null,
.build(); true,
new PermitAllAuthorizator(),
NO_OBSERVERS_INTERCEPTOR);
MqttConnectMessage connectMessage = MessageBuilder.connect().clientId(FAKE_PUBLISHER_ID)
.protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true).build();
m_processor.processConnect(m_channel, connectMessage); m_processor.processConnect(m_channel, connectMessage);
assertConnAckAccepted(m_channel); assertConnAckAccepted(m_channel);
MqttPublishMessage pubmsg = MessageBuilder.publish() MqttPublishMessage pubmsg = MessageBuilder.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.topicName(FAKE_TOPIC) .payload("Hello".getBytes()).retained(true).build();
.qos(MqttQoS.AT_MOST_ONCE)
.payload("Hello".getBytes())
.retained(true)
.build();
NettyUtils.clientID(m_channel, FAKE_PUBLISHER_ID); NettyUtils.clientID(m_channel, FAKE_PUBLISHER_ID);
m_processor.processPublish(m_channel, pubmsg); m_processor.processPublish(m_channel, pubmsg);
NettyUtils.cleanSession(m_channel, false); NettyUtils.cleanSession(m_channel, false);
//Exercise // Exercise
MqttSubscribeMessage msg = MessageBuilder.subscribe() MqttSubscribeMessage msg = MessageBuilder.subscribe().messageId(10).addSubscription(MqttQoS.AT_MOST_ONCE, "#")
.messageId(10)
.addSubscription(MqttQoS.AT_MOST_ONCE, "#")
.build(); .build();
m_processor.processSubscribe(m_channel, msg); m_processor.processSubscribe(m_channel, msg);
//Verify // Verify
//wait the latch // wait the latch
Object pubMessage = m_channel.readOutbound(); Object pubMessage = m_channel.readOutbound();
assertNotNull(pubMessage); assertNotNull(pubMessage);
assertTrue(pubMessage instanceof MqttPublishMessage); assertTrue(pubMessage instanceof MqttPublishMessage);
@@ -396,23 +385,28 @@ public class ProtocolProcessorTest {
retainedMessage.setClientID(FAKE_PUBLISHER_ID); retainedMessage.setClientID(FAKE_PUBLISHER_ID);
m_messagesStore.storePublishForFuture(retainedMessage); m_messagesStore.storePublishForFuture(retainedMessage);
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); m_processor.init(
subs,
m_messagesStore,
m_sessionStore,
null,
true,
new PermitAllAuthorizator(),
NO_OBSERVERS_INTERCEPTOR);
MqttConnectMessage connectMessage = MessageBuilder.connect() MqttConnectMessage connectMessage = MessageBuilder.connect().clientId(FAKE_PUBLISHER_ID)
.clientId(FAKE_PUBLISHER_ID) .protocolVersion(MqttVersion.MQTT_3_1).build();
.protocolVersion(MqttVersion.MQTT_3_1)
.build();
m_processor.processConnect(m_channel, connectMessage); m_processor.processConnect(m_channel, connectMessage);
//Verify no messages are still stored // Verify no messages are still stored
BlockingQueue<StoredMessage> messages = m_sessionStore.queue(FAKE_PUBLISHER_ID); BlockingQueue<StoredMessage> messages = m_sessionStore.queue(FAKE_PUBLISHER_ID);
assertTrue(messages.isEmpty()); assertTrue(messages.isEmpty());
} }
@Test @Test
public void publishNoPublishToInactiveSession() { public void publishNoPublishToInactiveSession() {
//create an inactive session for Subscriber // create an inactive session for Subscriber
m_sessionStore.createNewSession("Subscriber", false); m_sessionStore.createNewSession("Subscriber", false);
SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class);
@@ -420,69 +414,61 @@ public class ProtocolProcessorTest {
List<Subscription> inactiveSubscriptions = Collections.singletonList(inactiveSub); List<Subscription> inactiveSubscriptions = Collections.singletonList(inactiveSub);
when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions);
m_processor = new ProtocolProcessor(); m_processor = new ProtocolProcessor();
m_processor.init(mockedSubscriptions, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), m_processor.init(
mockedSubscriptions,
m_messagesStore,
m_sessionStore,
null,
true,
new PermitAllAuthorizator(),
NO_OBSERVERS_INTERCEPTOR); NO_OBSERVERS_INTERCEPTOR);
//Exercise // Exercise
MqttPublishMessage msg = MessageBuilder.publish() MqttPublishMessage msg = MessageBuilder.publish().topicName("/topic").qos(MqttQoS.AT_MOST_ONCE)
.topicName("/topic") .payload("Hello".getBytes()).retained(true).build();
.qos(MqttQoS.AT_MOST_ONCE)
.payload("Hello".getBytes())
.retained(true)
.build();
NettyUtils.clientID(m_channel, "Publisher"); NettyUtils.clientID(m_channel, "Publisher");
m_processor.processPublish(m_channel, msg); m_processor.processPublish(m_channel, msg);
//Verify no message is received // Verify no message is received
assertNull(m_channel.readOutbound()); assertNull(m_channel.readOutbound());
} }
/** /**
* Verify that receiving a publish with retained message and with Q0S = 0 * Verify that receiving a publish with retained message and with Q0S = 0 clean the existing
* clean the existing retained messages for that topic. * retained messages for that topic.
*/ */
@Test @Test
public void testCleanRetainedStoreAfterAQoS0AndRetainedTrue() { public void testCleanRetainedStoreAfterAQoS0AndRetainedTrue() {
//force a connect // force a connect
connMsg = MessageBuilder.connect() connMsg = MessageBuilder.connect().protocolVersion(MqttVersion.MQTT_3_1).clientId("Publisher")
.protocolVersion(MqttVersion.MQTT_3_1) .cleanSession(true).build();
.clientId("Publisher")
.cleanSession(true)
.build();
m_processor.processConnect(m_channel, connMsg); m_processor.processConnect(m_channel, connMsg);
//prepare and existing retained store // prepare and existing retained store
NettyUtils.clientID(m_channel, "Publisher"); NettyUtils.clientID(m_channel, "Publisher");
MqttPublishMessage msg = MessageBuilder.publish() MqttPublishMessage msg = MessageBuilder.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_LEAST_ONCE)
.topicName(FAKE_TOPIC) .payload("Hello".getBytes()).retained(true).messageId(100).build();
.qos(MqttQoS.AT_LEAST_ONCE)
.payload("Hello".getBytes())
.retained(true)
.messageId(100)
.build();
m_processor.processPublish(m_channel, msg); m_processor.processPublish(m_channel, msg);
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() { Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
public boolean match(String key) { public boolean match(String key) {
return SubscriptionsStore.matchTopics(key, FAKE_TOPIC); return SubscriptionsStore.matchTopics(key, FAKE_TOPIC);
} }
}); });
assertFalse(messages.isEmpty()); assertFalse(messages.isEmpty());
//Exercise // Exercise
MqttPublishMessage cleanPubMsg = MessageBuilder.publish() MqttPublishMessage cleanPubMsg = MessageBuilder.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE)
.topicName(FAKE_TOPIC) .payload("Hello".getBytes()).retained(true).build();
.qos(MqttQoS.AT_MOST_ONCE)
.payload("Hello".getBytes())
.retained(true)
.build();
m_processor.processPublish(m_channel, cleanPubMsg); m_processor.processPublish(m_channel, cleanPubMsg);
//Verify // Verify
messages = m_messagesStore.searchMatching(new IMatchingCondition() { messages = m_messagesStore.searchMatching(new IMatchingCondition() {
public boolean match(String key) { public boolean match(String key) {
return SubscriptionsStore.matchTopics(key, FAKE_TOPIC); return SubscriptionsStore.matchTopics(key, FAKE_TOPIC);
} }
}); });
assertTrue(messages.isEmpty()); assertTrue(messages.isEmpty());