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