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

On retained republish message adapt the QoS to the one specified during subscription (fix issue #411)

Browse Source
This commit is contained in:
andsel
2018-06-28 00:26:42 +02:00
parent 99226aadbd
commit 88e0ccf20e
6 changed files with 86 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java
+3 -3
View File
@@ -46,7 +46,7 @@ abstract class AbstractProtocolProcessorCommonUtils {
static final String TEST_USER = "fakeuser";
static final String TEST_PWD = "fakepwd";
private static final String HELLO_WORLD_MQTT = "Hello world MQTT!!";
static final String HELLO_WORLD_MQTT = "Hello world MQTT!!";
EmbeddedChannel m_channel;
ProtocolProcessor m_processor;
@@ -305,11 +305,11 @@ abstract class AbstractProtocolProcessorCommonUtils {
assertEquals("Connect must be accepted", CONNECTION_ACCEPTED, connAck.variableHeader().connectReturnCode());
}
protected void disconnect() throws InterruptedException {
protected void disconnect() {
disconnect(this.m_channel);
}
protected void disconnect(EmbeddedChannel channel) throws InterruptedException {
protected void disconnect(EmbeddedChannel channel) {
this.m_processor.processDisconnect(channel);
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java
+61 -32
View File
@@ -42,6 +42,7 @@ 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 io.netty.handler.codec.mqtt.MqttQoS.EXACTLY_ONCE;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
@@ -50,8 +51,9 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
static final String FAKE_CLIENT_ID = "FAKE_123";
static final String FAKE_CLIENT_ID2 = "FAKE_456";
static final String FAKE_PUBLISHER_ID = "Publisher";
static final String FAKE_TOPIC = "/news";
static final String PUBLISHER_ID = "Publisher";
private static final String SUBSCRIBER_ID = "Subscriber";
static final String NEWS_TOPIC = "/news";
static final String BAD_FORMATTED_TOPIC = "#MQTTClient";
static final String TEST_USER = "fakeuser";
@@ -69,17 +71,17 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
@Test
public void testPublishToItself() {
final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE);
final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(NEWS_TOPIC), AT_MOST_ONCE);
// subscriptions.matches(topic) redefine the method to return true
ISubscriptionsDirectory subs = new CTrieSubscriptionDirectory() {
@Override
public List<Subscription> matches(Topic topic) {
if (topic.toString().equals(FAKE_TOPIC)) {
if (topic.toString().equals(NEWS_TOPIC)) {
return Collections.singletonList(subscription);
} else {
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
throw new IllegalArgumentException("Expected " + NEWS_TOPIC + " buf found " + topic);
}
}
};
@@ -94,7 +96,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
connect_v3_1();
// Exercise
publishToAs("FakeCLI", FAKE_TOPIC, AT_MOST_ONCE, false);
publishToAs("FakeCLI", NEWS_TOPIC, AT_MOST_ONCE, false);
// Verify
verifyPublishIsReceived();
@@ -102,10 +104,10 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
@Test
public void testPublishToMultipleSubscribers() {
final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(FAKE_TOPIC), AT_MOST_ONCE);
final Subscription subscription = new Subscription(FAKE_CLIENT_ID, new Topic(NEWS_TOPIC), AT_MOST_ONCE);
final Subscription subscriptionClient2 = new Subscription(
FAKE_CLIENT_ID2,
new Topic(FAKE_TOPIC),
new Topic(NEWS_TOPIC),
AT_MOST_ONCE);
// subscriptions.matches(topic) redefine the method to return true
@@ -113,10 +115,10 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
@Override
public List<Subscription> matches(Topic topic) {
if (topic.toString().equals(FAKE_TOPIC)) {
if (topic.toString().equals(NEWS_TOPIC)) {
return Arrays.asList(subscription, subscriptionClient2);
} else {
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
throw new IllegalArgumentException("Expected " + NEWS_TOPIC + " buf found " + topic);
}
}
};
@@ -142,7 +144,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
assertConnAckAccepted(secondReceiverChannel);
// Exercise
MqttPublishMessage msg = MqttMessageBuilders.publish().topicName(FAKE_TOPIC).qos(AT_MOST_ONCE)
MqttPublishMessage msg = MqttMessageBuilders.publish().topicName(NEWS_TOPIC).qos(AT_MOST_ONCE)
.retained(false).payload(Unpooled.copiedBuffer("Hello".getBytes(UTF_8))).build();
NettyUtils.userName(m_channel, "FakeCLI");
m_processor.processPublish(m_channel, msg);
@@ -166,7 +168,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
connect();
// Exercise & verify
subscribe(FAKE_TOPIC, AT_MOST_ONCE);
subscribe(NEWS_TOPIC, AT_MOST_ONCE);
}
@Test
@@ -175,7 +177,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
NettyUtils.userName(m_channel, fakeUserName);
IAuthorizator mockAuthorizator = mock(IAuthorizator.class);
when(mockAuthorizator.canRead(eq(new Topic(FAKE_TOPIC)), eq(fakeUserName), eq(FAKE_CLIENT_ID)))
when(mockAuthorizator.canRead(eq(new Topic(NEWS_TOPIC)), eq(fakeUserName), eq(FAKE_CLIENT_ID)))
.thenReturn(false);
m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, mockAuthorizator,
@@ -184,7 +186,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
connect();
//Exercise
MqttSubAckMessage subAckMsg = subscribeWithoutVerify(FAKE_TOPIC, AT_MOST_ONCE);
MqttSubAckMessage subAckMsg = subscribeWithoutVerify(NEWS_TOPIC, AT_MOST_ONCE);
verifyFailureQos(subAckMsg);
}
@@ -199,11 +201,11 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
public void testDoubleSubscribe() {
connect();
assertEquals(0, subscriptions.size());
subscribe(FAKE_TOPIC, AT_MOST_ONCE);
subscribe(NEWS_TOPIC, AT_MOST_ONCE);
assertEquals(1, subscriptions.size());
//Exercise & verify
subscribe(FAKE_TOPIC, AT_MOST_ONCE);
subscribe(NEWS_TOPIC, AT_MOST_ONCE);
}
@Test
@@ -231,17 +233,17 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
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), AT_MOST_ONCE);
new Subscription(PUBLISHER_ID, new Topic(NEWS_TOPIC), AT_MOST_ONCE);
// subscriptions.matches(topic) redefine the method to return true
ISubscriptionsDirectory subs = new CTrieSubscriptionDirectory() {
@Override
public List<Subscription> matches(Topic topic) {
if (topic.toString().equals(FAKE_TOPIC)) {
if (topic.toString().equals(NEWS_TOPIC)) {
return Collections.singletonList(subscription);
} else {
throw new IllegalArgumentException("Expected " + FAKE_TOPIC + " buf found " + topic);
throw new IllegalArgumentException("Expected " + NEWS_TOPIC + " buf found " + topic);
}
}
};
@@ -252,8 +254,8 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
// 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, this.sessionsRepository);
connect_v3_1_asClient(FAKE_PUBLISHER_ID);
publishToAs(FAKE_PUBLISHER_ID, FAKE_TOPIC, AT_MOST_ONCE, true);
connect_v3_1_asClient(PUBLISHER_ID);
publishToAs(PUBLISHER_ID, NEWS_TOPIC, AT_MOST_ONCE, true);
NettyUtils.cleanSession(m_channel, false);
// Exercise
@@ -269,18 +271,18 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
List<Subscription> emptySubs = Collections.emptyList();
when(subs.matches(any(Topic.class))).thenReturn(emptySubs);
StoredMessage retainedMessage = new StoredMessage("Hello".getBytes(UTF_8), MqttQoS.EXACTLY_ONCE, "/topic");
StoredMessage retainedMessage = new StoredMessage("Hello".getBytes(UTF_8), EXACTLY_ONCE, "/topic");
retainedMessage.setRetained(true);
retainedMessage.setClientID(FAKE_PUBLISHER_ID);
retainedMessage.setClientID(PUBLISHER_ID);
m_messagesStore.storeRetained(new Topic("/topic"), retainedMessage);
m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(),
NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository);
connect_v3_1_asClient(FAKE_PUBLISHER_ID);
connect_v3_1_asClient(PUBLISHER_ID);
// Verify no messages are still stored
assertTrue(this.sessionsRepository.sessionForClient(FAKE_PUBLISHER_ID).isEmptyQueue());
assertTrue(this.sessionsRepository.sessionForClient(PUBLISHER_ID).isEmptyQueue());
}
@Test
@@ -313,27 +315,54 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils
connect_v3_1_asClient("Publisher");
// prepare and existing retained store
publishToAs("Publisher", FAKE_TOPIC, AT_LEAST_ONCE, 100, true);
publishToAs("Publisher", NEWS_TOPIC, AT_LEAST_ONCE, 100, true);
Collection<IMessagesStore.StoredMessage> messages = m_messagesStore
.searchMatching(key -> key.match(new Topic(FAKE_TOPIC)));
.searchMatching(key -> key.match(new Topic(NEWS_TOPIC)));
assertFalse(messages.isEmpty());
// Exercise
// send a message that clean the previous retained publish
publishToAs("Publisher", FAKE_TOPIC, AT_MOST_ONCE, true);
publishToAs("Publisher", NEWS_TOPIC, AT_MOST_ONCE, true);
// Verify
messages = m_messagesStore.searchMatching(key -> key.match(new Topic(FAKE_TOPIC)));
messages = m_messagesStore.searchMatching(key -> key.match(new Topic(NEWS_TOPIC)));
assertTrue(messages.isEmpty());
}
@Test
public void testLowerTheQosToTheRequestedBySubscription() {
Subscription subQos1 = new Subscription("Sub A", new Topic("a/b"), MqttQoS.AT_LEAST_ONCE);
assertEquals(MqttQoS.AT_LEAST_ONCE, lowerQosToTheSubscriptionDesired(subQos1, MqttQoS.EXACTLY_ONCE));
assertEquals(MqttQoS.AT_LEAST_ONCE, lowerQosToTheSubscriptionDesired(subQos1, EXACTLY_ONCE));
Subscription subQos2 = new Subscription("Sub B", new Topic("a/+"), MqttQoS.EXACTLY_ONCE);
assertEquals(MqttQoS.EXACTLY_ONCE, lowerQosToTheSubscriptionDesired(subQos2, MqttQoS.EXACTLY_ONCE));
Subscription subQos2 = new Subscription("Sub B", new Topic("a/+"), EXACTLY_ONCE);
assertEquals(EXACTLY_ONCE, lowerQosToTheSubscriptionDesired(subQos2, EXACTLY_ONCE));
}
@Test
public void testSubscribeReceivePublishedMessageAtSubscriberQoSAndNotPublisherQoS() {
ISubscriptionsDirectory subs = new CTrieSubscriptionDirectory();
MemoryStorageService storageService = new MemoryStorageService(null, null);
SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore(), null);
subs.init(sessionsRepository);
// 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, this.sessionsRepository);
connect_v3_1_asClient(PUBLISHER_ID);
publishQoS2ToAs(this.m_channel, PUBLISHER_ID, NEWS_TOPIC, 1, true);
disconnect();
// connect from the subscriber
EmbeddedChannel subscriberChannel = new EmbeddedChannel();
NettyUtils.clientID(subscriberChannel, SUBSCRIBER_ID);
NettyUtils.cleanSession(subscriberChannel, false);
connectAsClient(subscriberChannel, SUBSCRIBER_ID);
// subscribe to the topic news with granted QoS0
subscribe(subscriberChannel, NEWS_TOPIC, MqttQoS.AT_MOST_ONCE);
// Verify the retained message arrives with QoS0 instead of QoS2
verifyPublishIsReceived(subscriberChannel, HELLO_WORLD_MQTT, MqttQoS.AT_MOST_ONCE);
}
}
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java
+2 -2
View File
@@ -278,14 +278,14 @@ public class ProtocolProcessor_CONNECT_Test {
// subscribe
MqttSubscribeMessage subscribeMsg = MqttMessageBuilders.subscribe()
.addSubscription(MqttQoS.AT_MOST_ONCE, ProtocolProcessorTest.FAKE_TOPIC).messageId(10).build();
.addSubscription(MqttQoS.AT_MOST_ONCE, ProtocolProcessorTest.NEWS_TOPIC).messageId(10).build();
NettyUtils.clientID(m_session, FAKE_CLIENT_ID);
NettyUtils.cleanSession(m_session, false);
m_processor.processSubscribe(m_session, subscribeMsg);
Subscription expectedSubscription = new Subscription(
FAKE_CLIENT_ID,
new Topic(ProtocolProcessorTest.FAKE_TOPIC),
new Topic(ProtocolProcessorTest.NEWS_TOPIC),
MqttQoS.AT_MOST_ONCE);
verifySubscriptionExists(m_sessionStore, expectedSubscription);
assertEqualsSubAck(m_session.readOutbound());
broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java
+6 -6
View File
@@ -83,7 +83,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void testRetain_maintainMessage_againstClientDestruction() throws InterruptedException {
public void testRetain_maintainMessage_againstClientDestruction() {
connect();
publishToAs(FAKE_CLIENT_ID, "/topic", AT_LEAST_ONCE, 66, true);
disconnect();
@@ -135,7 +135,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void testPublishWithQoS1_notCleanSession() throws InterruptedException {
public void testPublishWithQoS1_notCleanSession() {
connect();
subscribe("/topic", AT_LEAST_ONCE);
disconnect();
@@ -148,21 +148,21 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
verifyPublishIsReceived();
}
protected void publishFromAnotherClient(String topic, MqttQoS qos) throws InterruptedException {
protected void publishFromAnotherClient(String topic, MqttQoS qos) {
EmbeddedChannel anotherChannel = new EmbeddedChannel();
connectAsClient(anotherChannel, "AnotherClient");
publishToAs(anotherChannel, "AnotherClient", topic, qos, 67, false);
disconnect(anotherChannel);
}
protected void publishFromAnotherClient(String topic, String payload, MqttQoS qos) throws InterruptedException {
protected void publishFromAnotherClient(String topic, String payload, MqttQoS qos) {
EmbeddedChannel anotherChannel = new EmbeddedChannel();
connectAsClient(anotherChannel, "AnotherClient");
publishToAs(anotherChannel, "AnotherClient", topic, payload, qos, 68, false);
disconnect(anotherChannel);
}
protected void publishQoS2FromAnotherClient(String topic) throws InterruptedException {
protected void publishQoS2FromAnotherClient(String topic) {
EmbeddedChannel anotherChannel = new EmbeddedChannel();
connectAsClient(anotherChannel, "AnotherClient");
publishQoS2ToAs(anotherChannel, "AnotherClient", topic, 67, false);
@@ -170,7 +170,7 @@ public class ProtocolProcessor_VariousUseCases_Test extends AbstractProtocolProc
}
@Test
public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws InterruptedException {
public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() {
connect();
subscribe("/topic", AT_LEAST_ONCE);
disconnect();