diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index a7b579d8..172a1e63 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -86,8 +86,6 @@ public class ClientSession { private final ISubscriptionsStore subscriptionsStore; private volatile boolean cleanSession; - // private BlockingQueue m_queueToPublish = new - // ArrayBlockingQueue<>(Constants.MAX_MESSAGE_QUEUE); private final OutboundFlightZone outboundFlightZone; private final InboundFlightZone inboundFlightZone; @@ -115,15 +113,15 @@ public class ClientSession { @Override public String toString() { - return "ClientSession{clientID='" + clientID + '\'' + "}"; + return "ClientSession{clientID='" + clientID + "'}"; } public boolean subscribe(Subscription newSubscription) { - LOG.info("Adding new subscription. ClientId={}, topics={}, qos={}", newSubscription.getClientId(), + LOG.info("Adding new subscription. CId={}, topics={}, qos={}", newSubscription.getClientId(), newSubscription.getTopicFilter(), newSubscription.getRequestedQos()); boolean validTopic = newSubscription.getTopicFilter().isValid(); if (!validTopic) { - LOG.warn("The topic filter is not valid. ClientId={}, topics={}", newSubscription.getClientId(), + LOG.warn("The topic filter is not valid. CId={}, topics={}", newSubscription.getClientId(), newSubscription.getTopicFilter()); // send SUBACK with 0x80 for this topic filter return false; @@ -133,7 +131,7 @@ public class ClientSession { // update the selected subscriptions if not present or if has a greater qos if (existingSub == null || existingSub.getRequestedQos().value() < newSubscription.getRequestedQos().value()) { if (existingSub != null) { - LOG.info("Subscription already existed with a lower QoS value. It will be updated. ClientId={}, " + + LOG.info("Subscription already existed with a lower QoS value. It will be updated. CId={}, " + "topics={}, existingQos={}, newQos={}", newSubscription.getClientId(), newSubscription.getTopicFilter(), existingSub.getRequestedQos(), newSubscription.getRequestedQos()); subscriptions.remove(newSubscription); @@ -145,7 +143,7 @@ public class ClientSession { } public void unsubscribeFrom(Topic topicFilter) { - LOG.info("Removing subscription. ClientID={}, topics={}", clientID, topicFilter); + LOG.info("Removing subscription. CId={}, topics={}", clientID, topicFilter); subscriptionsStore.removeSubscription(topicFilter, clientID); Set subscriptionsToRemove = new HashSet<>(); for (Subscription sub : this.subscriptions) { @@ -158,7 +156,7 @@ public class ClientSession { public void disconnect() { if (this.cleanSession) { - LOG.info("Client disconnected. Removing its subscriptions. ClientId={}", clientID); + LOG.info("Client disconnected. Removing its subscriptions. CId={}", clientID); // cleanup topic subscriptions cleanSession(); } @@ -191,7 +189,7 @@ public class ClientSession { * @return the packetID for the message in flight. * */ public int inFlightAckWaiting(IMessagesStore.StoredMessage msg) { - LOG.debug("Adding message ot inflight zone. ClientId={}", clientID); + LOG.debug("Adding message ot inflight zone. CId={}", clientID); int messageId = ClientSession.this.nextPacketId(); outboundFlightZone.waitingAck(messageId, msg); return messageId; @@ -224,6 +222,7 @@ public class ClientSession { } public Set getSubscriptions() { + //TODO WARN this doesn'load anything from DB return subscriptions; } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index cfee64f5..6acb919c 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -35,18 +35,19 @@ import io.netty.handler.codec.mqtt.*; import io.netty.handler.timeout.IdleStateHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import java.nio.ByteBuffer; import java.util.*; -import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; + +import static io.moquette.server.ConnectionDescriptor.ConnectionState.*; import static io.moquette.spi.impl.InternalRepublisher.createPublishForQos; import static io.moquette.spi.impl.Utils.messageId; import static io.moquette.spi.impl.Utils.readBytesAndRewind; import static io.netty.handler.codec.mqtt.MqttConnectReturnCode.*; import static io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader.from; import static io.netty.handler.codec.mqtt.MqttQoS.*; -import static io.moquette.server.ConnectionDescriptor.ConnectionState.*; /** * Class responsible to handle the logic of MQTT protocol it's the director of the protocol @@ -653,7 +654,7 @@ public class ProtocolProcessor { } if (!existingDescriptor.close()) { - LOG.info("The connection has been closed. CId={}", clientID); + LOG.info("Connection has been closed. CId={}", clientID); return; } @@ -664,7 +665,7 @@ public class ProtocolProcessor { return; } - LOG.info("The DISCONNECT message has been processed. CId={}", clientID); + LOG.info("DISCONNECT message has been processed. CId={}", clientID); } private boolean removeSubscriptions(ConnectionDescriptor descriptor, String clientID) { diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java deleted file mode 100644 index d13fc0ba..00000000 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationEmbeddedPublishTest.java +++ /dev/null @@ -1,295 +0,0 @@ -/* - * 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.server; - -import io.moquette.server.config.IConfig; -import io.moquette.server.config.MemoryConfig; -import io.netty.buffer.Unpooled; -import io.netty.handler.codec.mqtt.MqttMessageBuilders; -import io.netty.handler.codec.mqtt.MqttPublishMessage; -import io.netty.handler.codec.mqtt.MqttQoS; -import org.eclipse.paho.client.mqttv3.*; -import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; -import org.junit.After; -import org.junit.Before; -import org.junit.BeforeClass; -import org.junit.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import java.io.File; -import java.io.IOException; -import java.util.Properties; -import static org.junit.Assert.*; - -public class ServerIntegrationEmbeddedPublishTest { - - private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationEmbeddedPublishTest.class); - - static MqttClientPersistence s_dataStore; - static MqttClientPersistence s_pubDataStore; - - Server m_server; - IMqttClient m_subscriber; - MessageCollector m_callback; - IConfig m_config; - - protected void startServer() throws IOException { - m_server = new Server(); - final Properties configProps = IntegrationUtils.prepareTestProperties(); - m_config = new MemoryConfig(configProps); - m_server.startServer(m_config); - } - - @BeforeClass - public static void beforeTests() { - String tmpDir = System.getProperty("java.io.tmpdir"); - s_dataStore = new MqttDefaultFilePersistence(tmpDir); - s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher"); - } - - @Before - public void setUp() throws Exception { - startServer(); - - m_subscriber = new MqttClient("tcp://localhost:1883", "TestClient", s_dataStore); - m_callback = new MessageCollector(); - m_subscriber.setCallback(m_callback); - } - - @After - public void tearDown() throws Exception { - if (m_subscriber != null && m_subscriber.isConnected()) { - m_subscriber.disconnect(); - } - - m_server.stopServer(); - } - - private void subscribeToWithQos(String topic, int qos) throws Exception { - m_subscriber.connect(); - m_subscriber.subscribe(topic, qos); - } - - private void internalPublishToWithQosAndRetained(String topic, MqttQoS qos, boolean retained) { - MqttPublishMessage message = MqttMessageBuilders.publish().topicName(topic).retained(retained).qos(qos) - .payload(Unpooled.copiedBuffer("Hello world MQTT!!".getBytes())).build(); - m_server.internalPublish(message, "INTRPUBL"); - } - - private void connectNoCleanSession() throws Exception { - MqttConnectOptions opts = new MqttConnectOptions(); - opts.setCleanSession(false); - m_subscriber.connect(opts); - } - - private void subscribeToWithQosAndNoCleanSession(String topic, int qos) throws Exception { - connectNoCleanSession(); - m_subscriber.subscribe(topic, qos); - } - - private void verifyNoMessageIsReceived() throws Exception { - MqttMessage msg = m_callback.getMessageImmediate(); - assertNull(msg); - } - - private void verifyMessageIsReceivedSuccessfully() throws Exception { - // check in 2 seconds - MqttMessage msg = m_callback.waitMessage(2); - assertNotNull(msg); - assertEquals("Hello world MQTT!!", new String(msg.getPayload())); - } - - @Test - public void testClientSubscribeAfterNotRetainedQoS0IsSent() throws Exception { - LOG.info("*** testClientSubscribeAfterNotRetainedQoS0IsSent ***"); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.AT_MOST_ONCE, false); - subscribeToWithQos("/topic", 0); - - // Verify - verifyNoMessageIsReceived(); - } - - @Test - public void testClientSubscribeBeforeNotRetainedQoS0IsSent() throws Exception { - LOG.info("*** testClientSubscribeBeforeNotRetainedQoS0 ***"); - - subscribeToWithQos("/topic", 0); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.AT_MOST_ONCE, false); - - // Verify - verifyMessageIsReceivedSuccessfully(); - } - - @Test - public void testClientSubscribeBeforeRetainedQoS0IsSent() throws Exception { - LOG.info("*** testClientSubscribeBeforeRetainedQoS0IsSent ***"); - - subscribeToWithQos("/topic", 0); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.AT_MOST_ONCE, true); - - // Verify - verifyMessageIsReceivedSuccessfully(); - } - - @Test - public void testClientSubscribeAfterRetainedQoS0IsSent() throws Exception { - LOG.info("*** testClientSubscribeAfterRetainedQoS0IsSent ***"); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.AT_MOST_ONCE, true); - // LOG.info("** post internalPublish **"); - subscribeToWithQos("/topic", 0); - - // Verify - verifyNoMessageIsReceived(); - } - - @Test - public void testClientSubscribeBeforeNotRetainedQoS1IsSent() throws Exception { - LOG.info("*** testClientSubscribeBeforeNotRetainedQoS1IsSent ***"); - - subscribeToWithQos("/topic", 1); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.AT_LEAST_ONCE, false); - - // Verify - verifyMessageIsReceivedSuccessfully(); - } - - @Test - public void testClientSubscribeAfterNotRetainedQoS1IsSent() throws Exception { - LOG.info("*** testClientSubscribeAfterNotRetainedQoS1IsSent ***"); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.AT_LEAST_ONCE, false); - subscribeToWithQos("/topic", 1); - - // Verify - verifyNoMessageIsReceived(); - } - - @Test - public void testClientSubscribeBeforeRetainedQoS1IsSent() throws Exception { - LOG.info("*** testClientSubscribeBeforeRetainedQoS1IsSent ***"); - - subscribeToWithQos("/topic", 1); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.AT_LEAST_ONCE, true); - - // Verify - verifyMessageIsReceivedSuccessfully(); - } - - @Test - public void testClientSubscribeAfterRetainedQoS1IsSent() throws Exception { - LOG.info("*** testClientSubscribeAfterRetainedQoS0IsSent ***"); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.AT_LEAST_ONCE, true); - subscribeToWithQos("/topic", MqttQoS.AT_LEAST_ONCE.value()); - LOG.info("** After subscribe **"); - - // Verify - verifyMessageIsReceivedSuccessfully(); - LOG.info("** Post verify **"); - } - - @Test - public void testClientSubscribeBeforeNotRetainedQoS2IsSent() throws Exception { - LOG.info("*** testClientSubscribeBeforeNotRetainedQoS2IsSent ***"); - - subscribeToWithQos("/topic", 2); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.EXACTLY_ONCE, false); - - // Verify - verifyMessageIsReceivedSuccessfully(); - } - - @Test - public void testClientSubscribeAfterNotRetainedQoS2IsSent() throws Exception { - LOG.info("*** testClientSubscribeAfterNotRetainedQoS2IsSent ***"); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.EXACTLY_ONCE, false); - subscribeToWithQos("/topic", 2); - - // Verify - verifyNoMessageIsReceived(); - } - - @Test - public void testClientSubscribeBeforeRetainedQoS2IsSent() throws Exception { - LOG.info("*** testClientSubscribeBeforeRetainedQoS2IsSent ***"); - - subscribeToWithQos("/topic", 2); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.EXACTLY_ONCE, true); - - // Verify - verifyMessageIsReceivedSuccessfully(); - } - - @Test - public void testClientSubscribeAfterRetainedQoS2IsSent() throws Exception { - LOG.info("*** testClientSubscribeAfterRetainedQoS2IsSent ***"); - - // Exercise - internalPublishToWithQosAndRetained("/topic", MqttQoS.EXACTLY_ONCE, true); - subscribeToWithQos("/topic", 2); - - // Verify - verifyMessageIsReceivedSuccessfully(); - } - - @Test - public void testClientSubscribeAfterDisconnected() throws Exception { - LOG.info("*** testClientSubscribeAfterDisconnected ***"); - - subscribeToWithQos("foo", 0); - m_subscriber.disconnect(); - - internalPublishToWithQosAndRetained("foo", MqttQoS.AT_MOST_ONCE, false); - - // verifyMessageIsReceivedSuccessfully(); - // m_subscriber.receive(2, TimeUnit.MILLISECONDS); - MqttMessage message = m_callback.waitMessage(1); - assertTrue(message == null); - } - - @Test - public void testClientSubscribeWithoutCleanSession() throws Exception { - LOG.info("*** testClientSubscribeWithoutCleanSession ***"); - subscribeToWithQosAndNoCleanSession("foo", 1); - m_subscriber.disconnect(); - assertTrue(m_server.getSubscriptions().size() == 1); - connectNoCleanSession(); - assertTrue(m_server.getSubscriptions().size() == 1); - internalPublishToWithQosAndRetained("foo", MqttQoS.AT_MOST_ONCE, false); - verifyMessageIsReceivedSuccessfully(); - } -} diff --git a/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationRestartTest.java similarity index 99% rename from broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java rename to broker/src/test/java/io/moquette/server/ServerIntegrationRestartTest.java index 0abdd2bf..df6a5c9b 100644 --- a/broker/src/test/java/io/moquette/server/ServerRestartIntegrationTest.java +++ b/broker/src/test/java/io/moquette/server/ServerIntegrationRestartTest.java @@ -30,7 +30,7 @@ import java.util.Properties; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; -public class ServerRestartIntegrationTest { +public class ServerIntegrationRestartTest { static MqttClientPersistence s_dataStore; static MqttClientPersistence s_pubDataStore; diff --git a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java new file mode 100644 index 00000000..3903526b --- /dev/null +++ b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java @@ -0,0 +1,213 @@ +/* + * 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.interception.InterceptHandler; +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.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.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.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 { + + static final String FAKE_CLIENT_ID = "FAKE_123"; + static final String FAKE_CLIENT_ID2 = "FAKE_456"; + + static final String TEST_USER = "fakeuser"; + static final String TEST_PWD = "fakepwd"; + + static final List EMPTY_OBSERVERS = Collections.emptyList(); + static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); + public static final String HELLO_WORLD_MQTT = "Hello world MQTT!!"; + + EmbeddedChannel m_channel; + ProtocolProcessor m_processor; + + IMessagesStore m_messagesStore; + ISessionsStore m_sessionStore; + SubscriptionsDirectory subscriptions; + MockAuthenticator m_mockAuthenticator; + + protected void initializeProcessorAndSubsystems() { + 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 + MemoryStorageService memStorage = new MemoryStorageService(null, null); + m_messagesStore = memStorage.messagesStore(); + m_sessionStore = memStorage.sessionsStore(); + // m_messagesStore.initStore(); + + Set clientIds = new HashSet<>(); + clientIds.add(FAKE_CLIENT_ID); + clientIds.add(FAKE_CLIENT_ID2); + Map users = new HashMap<>(); + users.put(TEST_USER, TEST_PWD); + m_mockAuthenticator = new MockAuthenticator(clientIds, users); + + subscriptions = new SubscriptionsDirectory(); + subscriptions.init(memStorage.sessionsStore()); + m_processor = new ProtocolProcessor(); + m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + } + + protected void verifyNoPublishIsReceived() { + assertNull("No out messages from the processor", m_channel.readOutbound()); + } + + protected void subscribe(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(); + assertEquals(desiredQos.value(), (int) subAck.payload().grantedQoSLevels().get(0)); + + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, new Topic(topic), desiredQos); + assertTrue(subscriptions.contains(expectedSubscription)); + } + + protected MqttSubAckMessage subscribeWithoutVerify(String topic, MqttQoS desiredQos) { + MqttSubscribeMessage subscribe = MqttMessageBuilders.subscribe() + .addSubscription(desiredQos, topic) + .messageId(1) + .build(); + this.m_processor.processSubscribe(m_channel, subscribe); + return m_channel.readOutbound(); + } + + protected void subscribeAndNotReadResponse(String topic, MqttQoS desiredQos) { + MqttSubscribeMessage subscribe = MqttMessageBuilders.subscribe() + .addSubscription(desiredQos, topic) + .messageId(1) + .build(); + this.m_processor.processSubscribe(m_channel, subscribe); + } + + protected void unsubscribe(String topic) { + MqttUnsubscribeMessage msg = MqttMessageBuilders.unsubscribe() + .addTopicFilter(topic) + .messageId(1) + .build(); + + m_processor.processUnsubscribe(m_channel, msg); + } + + protected void internalPublishNotRetainedTo(String topic) { + internalPublishTo(topic, AT_MOST_ONCE, false); + } + + protected void internalPublishRetainedTo(String topic) { + internalPublishTo(topic, AT_MOST_ONCE, true); + } + + protected void internalPublishTo(String topic, MqttQoS qos, boolean retained) { + MqttPublishMessage publish = MqttMessageBuilders.publish() + .topicName(topic) + .retained(retained) + .qos(qos) + .payload(Unpooled.copiedBuffer(HELLO_WORLD_MQTT.getBytes())).build(); + this.m_processor.internalPublish(publish, "INTRPUBL"); + } + + protected void publishToAs(String clientId, String topic, MqttQoS qos, boolean retained) { + NettyUtils.userName(m_channel, clientId); + MqttPublishMessage publish = MqttMessageBuilders.publish() + .topicName(topic) + .retained(retained) + .qos(qos) + .payload(Unpooled.copiedBuffer(HELLO_WORLD_MQTT.getBytes())).build(); + this.m_processor.processPublish(m_channel, publish); + } + + protected void publishToAs(String clientId, String topic, MqttQoS qos, int messageId, boolean retained) { + NettyUtils.userName(m_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); + } + + + protected void connect() { + connectAsClient(FAKE_CLIENT_ID); + } + + protected void connectAsClient(String clientId) { + MqttConnectMessage connectMessage = MqttMessageBuilders.connect() + .clientId(clientId) + .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 connect_v3_1() { + connect_v3_1_asClient(FAKE_CLIENT_ID); + } + + protected void connect_v3_1_asClient(String clientID) { + MqttConnectMessage connectMessage = MqttMessageBuilders.connect() + .clientId(clientID) + .cleanSession(true) + .protocolVersion(MqttVersion.MQTT_3_1) + .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 connectNoCleanSession() { + MqttConnectMessage connectMessage = MqttMessageBuilders.connect() + .clientId(FAKE_CLIENT_ID) + .cleanSession(false) + .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); + } + + protected void verifyPublishIsReceived() { + final MqttPublishMessage publishReceived = m_channel.readOutbound(); + String payloadMessage = new String(publishReceived.payload().array()); + assertEquals("Sent and received payload must be identical", HELLO_WORLD_MQTT, payloadMessage); + } +} 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 59f787ca..86efb16b 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -21,7 +21,6 @@ import io.moquette.persistence.MemoryStorageService; import io.moquette.server.netty.NettyUtils; import io.moquette.spi.IMessagesStore; import io.moquette.spi.IMessagesStore.StoredMessage; -import io.moquette.spi.ISessionsStore; import io.moquette.spi.impl.security.PermitAllAuthorizator; import io.moquette.spi.impl.subscriptions.Subscription; import io.moquette.spi.impl.subscriptions.SubscriptionsDirectory; @@ -32,14 +31,17 @@ 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 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.*; import static org.mockito.Mockito.*; -public class ProtocolProcessorTest { +public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils { static final String FAKE_CLIENT_ID = "FAKE_123"; static final String FAKE_CLIENT_ID2 = "FAKE_456"; @@ -55,52 +57,15 @@ public class ProtocolProcessorTest { static final List EMPTY_OBSERVERS = Collections.emptyList(); static final BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); - EmbeddedChannel m_channel; - MqttConnectMessage connMsg; - ProtocolProcessor m_processor; - - IMessagesStore m_messagesStore; - ISessionsStore m_sessionStore; - SubscriptionsDirectory subscriptions; - MockAuthenticator m_mockAuthenticator; @Before public void setUp() throws InterruptedException { - /* - * connMsg = new ConnectMessage(); connMsg.setProtocolVersion((byte) 0x03); - * - * connMsg = MqttMessageBuilders.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 - Thread.sleep(300); - MemoryStorageService memStorage = new MemoryStorageService(null, null); - m_messagesStore = memStorage.messagesStore(); - m_sessionStore = memStorage.sessionsStore(); - // m_messagesStore.initStore(); - - Set clientIds = new HashSet<>(); - clientIds.add(FAKE_CLIENT_ID); - clientIds.add(FAKE_CLIENT_ID2); - Map users = new HashMap<>(); - users.put(TEST_USER, TEST_PWD); - m_mockAuthenticator = new MockAuthenticator(clientIds, users); - - subscriptions = new SubscriptionsDirectory(); - subscriptions.init(memStorage.sessionsStore()); - m_processor = new ProtocolProcessor(); - m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); + initializeProcessorAndSubsystems(); } @Test public void testPublishToItself() throws InterruptedException { - final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), MqttQoS.AT_MOST_ONCE); + final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true SubscriptionsDirectory subs = new SubscriptionsDirectory() { @@ -118,37 +83,25 @@ public class ProtocolProcessorTest { // simulate a connect that register a clientID to an IoSession MemoryStorageService storageService = new MemoryStorageService(null, null); subs.init(storageService.sessionsStore()); - m_processor.init( - subs, - m_messagesStore, - m_sessionStore, - null, - true, - new PermitAllAuthorizator(), + m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); - MqttConnectMessage connectMessage = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1) - .clientId(FAKE_CLIENT_ID).cleanSession(true).build(); - m_processor.processConnect(m_channel, connectMessage); + + connect_v3_1(); // Exercise - MqttPublishMessage msg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE) - .retained(false).payload(Unpooled.copiedBuffer("Hello".getBytes())).build(); - - NettyUtils.userName(m_channel, "FakeCLI"); - m_processor.processPublish(m_channel, msg); + publishToAs("FakeCLI", FAKE_TOPIC, AT_MOST_ONCE, false); // Verify - assertNotNull(m_channel.readOutbound()); - // TODO check received message attributes + verifyPublishIsReceived(); } @Test public void testPublishToMultipleSubscribers() throws InterruptedException { - final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), MqttQoS.AT_MOST_ONCE); + final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE); final Subscription subscriptionClient2 = new Subscription( FAKE_CLIENT_ID2, new Topic(FAKE_TOPIC), - MqttQoS.AT_MOST_ONCE); + AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true SubscriptionsDirectory subs = new SubscriptionsDirectory() { @@ -183,7 +136,7 @@ public class ProtocolProcessorTest { assertConnAckAccepted(secondReceiverChannel); // Exercise - MqttPublishMessage msg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE) + MqttPublishMessage msg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(AT_MOST_ONCE) .retained(false).payload(Unpooled.copiedBuffer("Hello".getBytes())).build(); NettyUtils.userName(m_channel, "FakeCLI"); m_processor.processPublish(m_channel, msg); @@ -204,17 +157,10 @@ public class ProtocolProcessorTest { @Test public void testSubscribe() { - MqttSubscribeMessage msg = MqttMessageBuilders.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); + connect(); - // Verify - assertTrue(m_channel.readOutbound() instanceof MqttSubAckMessage); - Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), - MqttQoS.AT_MOST_ONCE); - assertTrue(subscriptions.contains(expectedSubscription)); + // Exercise & verify + subscribe(FAKE_TOPIC, AT_MOST_ONCE); } @Test @@ -229,17 +175,11 @@ public class ProtocolProcessorTest { m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, mockAuthorizator, NO_OBSERVERS_INTERCEPTOR); - // Exercise - MqttSubscribeMessage msg = MqttMessageBuilders.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC) - .messageId(10).build(); + connect(); - m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); - m_processor.processSubscribe(m_channel, msg); + //Exercise + MqttSubAckMessage subAckMsg = subscribeWithoutVerify(FAKE_TOPIC, AT_MOST_ONCE); - // Verify - Object ackMsg = m_channel.readOutbound(); - assertTrue(ackMsg instanceof MqttSubAckMessage); - MqttSubAckMessage subAckMsg = (MqttSubAckMessage) ackMsg; verifyFailureQos(subAckMsg); } @@ -251,63 +191,41 @@ public class ProtocolProcessorTest { @Test public void testDoubleSubscribe() { - MqttSubscribeMessage msg = MqttMessageBuilders.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, FAKE_TOPIC) - .messageId(10).build(); - m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); + connect(); assertEquals(0, subscriptions.size()); - - m_processor.processSubscribe(m_channel, msg); + subscribe(FAKE_TOPIC, AT_MOST_ONCE); assertEquals(1, subscriptions.size()); - //Exercise - m_processor.processSubscribe(m_channel, msg); - - // Verify - assertEquals(1, subscriptions.size()); - Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), - MqttQoS.AT_MOST_ONCE); - - assertTrue(subscriptions.contains(expectedSubscription)); + //Exercise & verify + subscribe(FAKE_TOPIC, AT_MOST_ONCE); } @Test public void testSubscribeWithBadFormattedTopic() { - MqttSubscribeMessage msg = MqttMessageBuilders.subscribe().addSubscription(MqttQoS.AT_MOST_ONCE, BAD_FORMATTED_TOPIC) - .messageId(10).build(); - - m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); + connect(); assertEquals(0, subscriptions.size()); - // Exercise - m_processor.processSubscribe(m_channel, msg); + //Exercise + MqttSubAckMessage subAckMsg = subscribeWithoutVerify(BAD_FORMATTED_TOPIC, AT_MOST_ONCE); - // Verify assertEquals(0, subscriptions.size()); - Object recvSubAckMessage = m_channel.readOutbound(); - assertTrue(recvSubAckMessage instanceof MqttSubAckMessage); - verifyFailureQos((MqttSubAckMessage) recvSubAckMessage); + verifyFailureQos(subAckMsg); } - /* - * Check topicFilter is a valid MQTT topic filter (issue 68) - */ @Test public void testUnsubscribeWithBadFormattedTopic() { - MqttUnsubscribeMessage msg = MqttMessageBuilders.unsubscribe().addTopicFilter(BAD_FORMATTED_TOPIC).messageId(1) - .build(); - // Exercise - m_processor.processUnsubscribe(m_channel, msg); + unsubscribe(BAD_FORMATTED_TOPIC); // 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, (issue 68)", 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, new Topic(FAKE_TOPIC), MqttQoS.AT_MOST_ONCE); + new Subscription(FAKE_PUBLISHER_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE); // subscriptions.matches(topic) redefine the method to return true SubscriptionsDirectory subs = new SubscriptionsDirectory() { @@ -327,29 +245,15 @@ public class ProtocolProcessorTest { // 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 = MqttMessageBuilders.connect().clientId(FAKE_PUBLISHER_ID) - .protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true).build(); - - m_processor.processConnect(m_channel, connectMessage); - assertConnAckAccepted(m_channel); - MqttPublishMessage pubmsg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE) - .payload(Unpooled.copiedBuffer("Hello".getBytes())).retained(true).build(); - - NettyUtils.clientID(m_channel, FAKE_PUBLISHER_ID); - m_processor.processPublish(m_channel, pubmsg); + connect_v3_1_asClient(FAKE_PUBLISHER_ID); + publishToAs(FAKE_PUBLISHER_ID, FAKE_TOPIC, AT_MOST_ONCE, true); NettyUtils.cleanSession(m_channel, false); // Exercise - MqttSubscribeMessage msg = MqttMessageBuilders.subscribe().messageId(10).addSubscription(MqttQoS.AT_MOST_ONCE, "#") - .build(); - m_processor.processSubscribe(m_channel, msg); + subscribeAndNotReadResponse("#", AT_MOST_ONCE); // Verify - // wait the latch - Object pubMessage = m_channel.readOutbound(); - assertNotNull(pubMessage); - assertTrue(pubMessage instanceof MqttPublishMessage); - assertEquals(FAKE_TOPIC, ((MqttPublishMessage) pubMessage).variableHeader().topicName()); + verifyPublishIsReceived(); } @Test @@ -366,10 +270,7 @@ public class ProtocolProcessorTest { m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); - MqttConnectMessage connectMessage = MqttMessageBuilders.connect().clientId(FAKE_PUBLISHER_ID) - .protocolVersion(MqttVersion.MQTT_3_1).build(); - - m_processor.processConnect(m_channel, connectMessage); + connect_v3_1_asClient(FAKE_PUBLISHER_ID); // Verify no messages are still stored Queue messages = m_sessionStore.queue(FAKE_PUBLISHER_ID); @@ -390,14 +291,10 @@ public class ProtocolProcessorTest { NO_OBSERVERS_INTERCEPTOR); // Exercise - MqttPublishMessage msg = MqttMessageBuilders.publish().topicName("/topic").qos(MqttQoS.AT_MOST_ONCE) - .payload(Unpooled.copiedBuffer("Hello".getBytes())).retained(true).build(); + connectAsClient("Publisher"); + publishToAs("Publisher", "/topic", AT_MOST_ONCE, true); - NettyUtils.clientID(m_channel, "Publisher"); - m_processor.processPublish(m_channel, msg); - - // Verify no message is received - assertNull(m_channel.readOutbound()); + verifyNoPublishIsReceived(); } /** @@ -407,24 +304,18 @@ public class ProtocolProcessorTest { @Test public void testCleanRetainedStoreAfterAQoS0AndRetainedTrue() { // force a connect - connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).clientId("Publisher") - .cleanSession(true).build(); - m_processor.processConnect(m_channel, connMsg); + connect_v3_1_asClient("Publisher"); + // prepare and existing retained store - NettyUtils.clientID(m_channel, "Publisher"); - MqttPublishMessage msg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_LEAST_ONCE) - .payload(Unpooled.copiedBuffer("Hello".getBytes())).retained(true).messageId(100).build(); - m_processor.processPublish(m_channel, msg); + publishToAs("Publisher", FAKE_TOPIC, AT_LEAST_ONCE, 100, true); Collection messages = m_messagesStore .searchMatching(key -> key.match(new Topic(FAKE_TOPIC))); assertFalse(messages.isEmpty()); // Exercise - MqttPublishMessage cleanPubMsg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(MqttQoS.AT_MOST_ONCE) - .payload(Unpooled.copiedBuffer("Hello".getBytes())).retained(true).build(); - - m_processor.processPublish(m_channel, cleanPubMsg); + // send a message that clean the previous retained publish + publishToAs("Publisher", FAKE_TOPIC, AT_MOST_ONCE, true); // Verify messages = m_messagesStore.searchMatching(key -> key.match(new Topic(FAKE_TOPIC))); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 0c24b929..4c94d13c 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -58,8 +58,6 @@ public class ProtocolProcessor_CONNECT_Test { m_session = new EmbeddedChannel(); - // sleep to let the messaging batch processor to process the initEvent - Thread.sleep(300); MemoryStorageService memStorage = new MemoryStorageService(null, null); m_messagesStore = memStorage.messagesStore(); m_sessionStore = memStorage.sessionsStore(); @@ -84,18 +82,6 @@ public class ProtocolProcessor_CONNECT_Test { ProtocolProcessorTest.NO_OBSERVERS_INTERCEPTOR); } - // @Test - // public void testHandleConnect_BadProtocol() { - // connMsg.setProtocolVersion((byte) 0x02); - // - // //Exercise - // m_processor.processConnect(m_session, connMsg); - // - // //Verify - // assertEqualsConnAck(UNNACEPTABLE_PROTOCOL_VERSION, m_session.readOutbound()); - // assertFalse("Connection should be closed by the broker.", m_session.isOpen()); - // } - @Test public void testConnect_badClientID() { connMsg.clientId("extremely_long_clientID_greater_than_23").build(); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_InternalPublish_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_InternalPublish_Test.java new file mode 100644 index 00000000..d6dff32f --- /dev/null +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_InternalPublish_Test.java @@ -0,0 +1,191 @@ +/* + * 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.netty.NettyUtils; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.mqtt.MqttQoS; +import org.junit.Before; +import org.junit.Test; + +import static io.netty.handler.codec.mqtt.MqttQoS.*; +import static org.junit.Assert.assertEquals; + +public class ProtocolProcessor_InternalPublish_Test extends AbstractProtocolProcessorCommonUtils { + + @Before + public void setUp() throws InterruptedException { + initializeProcessorAndSubsystems(); + connect(); + } + + @Test + public void testClientSubscribeAfterNotRetainedQoS0IsSent() { + // Exercise + final String topic = "/topic"; + internalPublishNotRetainedTo(topic); + + subscribe(topic, AT_MOST_ONCE); + + // Verify + verifyNoPublishIsReceived(); + } + + @Test + public void testClientSubscribeAfterRetainedQoS0IsSent() { + // Exercise + final String topic = "/topic"; + internalPublishRetainedTo(topic); + + subscribe(topic, AT_MOST_ONCE); + + // Verify + verifyNoPublishIsReceived(); + } + + @Test + public void testClientSubscribeBeforeNotRetainedQoS0IsSent() { + subscribe("/topic", AT_MOST_ONCE); + + // Exercise + internalPublishNotRetainedTo("/topic"); + + // Verify + //verifyMessageIsReceivedSuccessfully(); + verifyPublishIsReceived(); + } + + @Test + public void testClientSubscribeBeforeRetainedQoS0IsSent() { + subscribe("/topic", AT_MOST_ONCE); + + // Exercise + internalPublishRetainedTo("/topic"); + + // Verify + verifyPublishIsReceived(); + } + + @Test + public void testClientSubscribeBeforeNotRetainedQoS1IsSent() { + subscribe("/topic", AT_LEAST_ONCE); + + // Exercise + internalPublishTo("/topic", AT_LEAST_ONCE, false); + + // Verify + verifyPublishIsReceived(); + } + + @Test + public void testClientSubscribeAfterNotRetainedQoS1IsSent() { + // Exercise + internalPublishTo("/topic", AT_LEAST_ONCE, false); + subscribe("/topic", AT_LEAST_ONCE); + + // Verify + verifyNoPublishIsReceived(); + } + + @Test + public void testClientSubscribeBeforeRetainedQoS1IsSent() { + subscribe("/topic", AT_LEAST_ONCE); + + // Exercise + internalPublishTo("/topic", AT_LEAST_ONCE, true); + + // Verify + verifyPublishIsReceived(); + } + + @Test + public void testClientSubscribeAfterRetainedQoS1IsSent() { + // Exercise + internalPublishTo("/topic", AT_LEAST_ONCE, true); + subscribe("/topic", AT_LEAST_ONCE); + + // Verify + verifyPublishIsReceived(); + } + + @Test + public void testClientSubscribeBeforeNotRetainedQoS2IsSent() { + subscribe("/topic", EXACTLY_ONCE); + + // Exercise + internalPublishTo("/topic", EXACTLY_ONCE, false); + + // Verify + verifyPublishIsReceived(); + } + + @Test + public void testClientSubscribeAfterNotRetainedQoS2IsSent() { + // Exercise + internalPublishTo("/topic", EXACTLY_ONCE, false); + subscribe("/topic", EXACTLY_ONCE); + + // Verify + verifyNoPublishIsReceived(); + } + + @Test + public void testClientSubscribeBeforeRetainedQoS2IsSent() { + subscribe("/topic", EXACTLY_ONCE); + + // Exercise + internalPublishTo("/topic", EXACTLY_ONCE, true); + + // Verify + verifyPublishIsReceived(); + } + + @Test + public void testClientSubscribeAfterRetainedQoS2IsSent() { + // Exercise + internalPublishTo("/topic", EXACTLY_ONCE, true); + subscribe("/topic", EXACTLY_ONCE); + + // Verify + verifyPublishIsReceived(); + } + + @Test + public void testClientSubscribeAfterDisconnected() throws InterruptedException { + subscribe("foo", AT_MOST_ONCE); + disconnect(); + + internalPublishTo("foo", AT_MOST_ONCE, false); + + verifyNoPublishIsReceived(); + } + + @Test + public void testClientSubscribeWithoutCleanSession() throws Exception { + subscribe("foo", AT_MOST_ONCE); + disconnect(); + assertEquals(1, m_processor.getSessionsStore().subscriptionStore().getSubscriptions().size()); + + m_channel = new EmbeddedChannel(); + NettyUtils.clientID(m_channel, FAKE_CLIENT_ID); + + connectNoCleanSession(); + assertEquals(1, m_processor.getSessionsStore().subscriptionStore().getSubscriptions().size()); + internalPublishTo("foo", MqttQoS.AT_MOST_ONCE, false); + verifyPublishIsReceived(); + } + +}