From 72c8445693221cd0cc39e6acd55ded30da17b2ce Mon Sep 17 00:00:00 2001 From: andrea Date: Tue, 27 Jan 2015 00:15:14 +0100 Subject: [PATCH] Fixed issue #12, missed remove of already published retained messages --- .../spi/impl/MemoryStorageService.java | 2 +- .../moquette/spi/impl/ProtocolProcessor.java | 1 + .../spi/persistence/MapDBPersistentStore.java | 1 + .../spi/impl/ProtocolProcessorTest.java | 27 ++++-- .../issue_12_multiple_republish_qos2.groovy | 89 +++++++++++++++++++ 5 files changed, 114 insertions(+), 6 deletions(-) create mode 100644 tools_scripts/issue_12_multiple_republish_qos2.groovy diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index c163ccf5..f8b98312 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -94,7 +94,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { @Override public List retrievePersistedPublishes(String clientID) { - return m_persistentMessageStore.get(clientID); + return new ArrayList<>(m_persistentMessageStore.get(clientID)); } @Override diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 7c9cddf8..d45fc3f1 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -248,6 +248,7 @@ class ProtocolProcessor implements EventHandler { for (PublishEvent pubEvt : publishedEvents) { sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(), pubEvt.getMessage(), false, pubEvt.getMessageID()); + m_messagesStore.cleanPersistedPublishMessage(clientID, pubEvt.getMessageID()); } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java index 62ea5278..f2c965e0 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java @@ -237,6 +237,7 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { } public void removeQoS2Message(String publishKey) { + LOG.debug("Removing stored Q0S2 message <{}>", publishKey); m_qos2Store.remove(publishKey); } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index ae9f1959..756d07ce 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -16,17 +16,14 @@ package org.eclipse.moquette.spi.impl; import java.nio.ByteBuffer; -import java.util.Arrays; -import java.util.Collection; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; import org.eclipse.moquette.spi.impl.events.LostConnectionEvent; +import org.eclipse.moquette.spi.impl.events.PublishEvent; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore; import static org.eclipse.moquette.parser.netty.Utils.VERSION_3_1_1; @@ -489,6 +486,26 @@ public class ProtocolProcessorTest { PublishMessage pubMessage = (PublishMessage) m_receivedMessage; assertEquals(FAKE_TOPIC, pubMessage.getTopicName()); } + + @Test + public void testRepublishAndConsumePersistedMessages_onReconnect() { + SubscriptionsStore subs = mock(SubscriptionsStore.class); + List emptySubs = Collections.emptyList(); + when(subs.matches(anyString())).thenReturn(emptySubs); + PublishEvent retainedMessage = new PublishEvent("/topic", QOSType.EXACTLY_ONCE, + ByteBuffer.wrap("Hello".getBytes()), true, FAKE_PUBLISHER_ID, 120); + m_storageService.storePublishForFuture(retainedMessage); + + m_processor.init(subs, m_storageService, m_sessionStore, null); + ConnectMessage connectMessage = new ConnectMessage(); + connectMessage.setClientID(FAKE_PUBLISHER_ID); + connectMessage.setProcotolVersion((byte)3); + connectMessage.setCleanSession(false); + m_processor.processConnect(m_session, connectMessage); + + //Verify no messages are still stored + assertTrue(m_storageService.retrievePersistedPublishes(FAKE_PUBLISHER_ID).isEmpty()); + } @Test public void publishNoPublishToInactiveSubscriptions() { diff --git a/tools_scripts/issue_12_multiple_republish_qos2.groovy b/tools_scripts/issue_12_multiple_republish_qos2.groovy new file mode 100644 index 00000000..99700437 --- /dev/null +++ b/tools_scripts/issue_12_multiple_republish_qos2.groovy @@ -0,0 +1,89 @@ +@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5') + +import org.fusesource.mqtt.client.BlockingConnection +import org.fusesource.mqtt.client.MQTT +import org.fusesource.mqtt.client.Message +import org.fusesource.mqtt.client.QoS +import org.fusesource.mqtt.client.Topic + +//test case moquette 0.7 update for 2015-1-26 +//client eclipse.paht.javaclient version :1.0.1 +String host = args.size() > 0 ? args[0] : 'localhost' + +MQTT mqttPub = new MQTT() +mqttPub.setHost(host, 1883) +mqttPub.setCleanSession(false) +mqttPub.setClientId("PublisherClient") + +MQTT mqttSub = new MQTT() +mqttSub.setHost(host, 1883) +mqttSub.setCleanSession(false) +mqttSub.setClientId("FlapSubscriberClient") + +BlockingConnection publisher = mqttPub.blockingConnection() +publisher.connect() +println "Publisher connected to ${host}" + +BlockingConnection subscriber = mqttSub.blockingConnection() +subscriber.connect() +println "${subscriber} connected to ${host}" +Topic[] topics = [new Topic("/topic", QoS.AT_MOST_ONCE)] +byte[] qoses = subscriber.subscribe(topics) +println "${subscriber} Subscribed to ${topics}" + +subscriber.disconnect() +println "${subscriber} disconnects immediately" + +println "${publisher} publishes QoS 2 messages on /topic" +publisher.publish("/topic", 'Hello world!!'.bytes, QoS.EXACTLY_ONCE, false) + + +subscriber = mqttSub.blockingConnection() +subscriber.connect() +println "${subscriber} reconnects to ${host}" + +Message message = subscriber.receive() +assert "/topic" == message.topic +String msgBody = new String(message.payload) +assert "Hello world!!" == msgBody +message.ack() +println "${subscriber} received on topic ${message.topic} the message \n\t ${msgBody}" +subscriber.disconnect() +println "${subscriber} disconnects again" + +subscriber = mqttSub.blockingConnection() +subscriber.connect() +println "${subscriber} reconnects again to ${host}" + +println "Wait receive something..." +message = subscriber.receive() +msgBody = new String(message.payload) +assert "Hello world!!" == msgBody +message.ack() +println "${subscriber} received again on topic ${message.topic} the message \n\t ${msgBody}" + +//aclient (clear session false) +//bclient (clear session false) +// +//a -> connection Server +//b -> connection Server +//b -> subscriptions (qos =0) +//b -> distconnection or lostconnection +// +//a ->publish ("abc/topic","test1",qos=2,false); +// +//... +//a ->publish ("abc/topic","test10",qos=2,false); +// +//b -> connection Server +//received message test1 +//... +//received message test10 +// +//b-> distconnection Server or loast connection +//b->connnection Server +//received message test1 +//... +//received message test10 +// +//why repeat received message