From 88e0ccf20e48dacc7528b7e684a255483f55d8ed Mon Sep 17 00:00:00 2001 From: andsel Date: Thu, 28 Jun 2018 00:26:42 +0200 Subject: [PATCH] On retained republish message adapt the QoS to the one specified during subscription (fix issue #411) --- .../java/io/moquette/spi/IMessagesStore.java | 7 +- .../moquette/spi/impl/ProtocolProcessor.java | 8 ++ .../AbstractProtocolProcessorCommonUtils.java | 6 +- .../spi/impl/ProtocolProcessorTest.java | 93 ++++++++++++------- .../impl/ProtocolProcessor_CONNECT_Test.java | 4 +- ...rotocolProcessor_VariousUseCases_Test.java | 12 +-- 6 files changed, 86 insertions(+), 44 deletions(-) diff --git a/broker/src/main/java/io/moquette/spi/IMessagesStore.java b/broker/src/main/java/io/moquette/spi/IMessagesStore.java index da5c38ef..209834a2 100644 --- a/broker/src/main/java/io/moquette/spi/IMessagesStore.java +++ b/broker/src/main/java/io/moquette/spi/IMessagesStore.java @@ -31,7 +31,7 @@ public interface IMessagesStore { class StoredMessage implements Serializable { private static final long serialVersionUID = 1755296138639817304L; - final MqttQoS m_qos; + private MqttQoS m_qos; final byte[] m_payload; final String m_topic; private boolean m_retained; @@ -44,6 +44,11 @@ public interface IMessagesStore { m_topic = topic; } + + public void setQos(MqttQoS qos) { + this.m_qos = qos; + } + public MqttQoS getQos() { return m_qos; } 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 abc78ca1..7a9f616f 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -858,6 +858,14 @@ public class ProtocolProcessor { newSubscription.getClientId(), newSubscription.getTopicFilter(), messages.size()); } ClientSession targetSession = this.sessionsRepository.sessionForClient(newSubscription.getClientId()); + + // adapt the message QoS to subscriber's accepted accepted + for (IMessagesStore.StoredMessage msg : messages) { + int lowestQoS = Math.min(msg.getQos().value(), newSubscription.getRequestedQos().value()); + MqttQoS qosToPublish = MqttQoS.valueOf(lowestQoS); + msg.setQos(qosToPublish); + } + this.internalRepublisher.publishRetained(targetSession, messages); // notify the Observables diff --git a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java index 164a39e9..63633e60 100644 --- a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java +++ b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java @@ -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); } 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 0cd03738..7fb06d3b 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -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 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 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 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 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 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); } } 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 be92728c..e2f1c5e0 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 @@ -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()); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java index 4ff3965b..de92dde0 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_VariousUseCases_Test.java @@ -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();