Moved tests from integration to unit and extracted and simplified common test part in ProtocolProcessor unit test
This commit is contained in:
andsel
@@ -86,8 +86,6 @@ public class ClientSession {
|
||||
private final ISubscriptionsStore subscriptionsStore;
|
||||
private volatile boolean cleanSession;
|
||||
|
||||
// private BlockingQueue<AbstractMessage> 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<Subscription> 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<Subscription> getSubscriptions() {
|
||||
//TODO WARN this doesn'load anything from DB
|
||||
return subscriptions;
|
||||
}
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -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;
|
||||
@@ -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<InterceptHandler> 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<String> clientIds = new HashSet<>();
|
||||
clientIds.add(FAKE_CLIENT_ID);
|
||||
clientIds.add(FAKE_CLIENT_ID2);
|
||||
Map<String, String> 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);
|
||||
}
|
||||
}
|
||||
@@ -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<InterceptHandler> 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<String> clientIds = new HashSet<>();
|
||||
clientIds.add(FAKE_CLIENT_ID);
|
||||
clientIds.add(FAKE_CLIENT_ID2);
|
||||
Map<String, String> 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<StoredMessage> 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<IMessagesStore.StoredMessage> 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)));
|
||||
|
||||
@@ -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();
|
||||
|
||||
+191
@@ -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();
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user