Fixed issue #12, missed remove of already published retained messages

2015-01-27 00:15:14 +01:00
parent 5fe9554965
commit 72c8445693
5 changed files with 114 additions and 6 deletions
@@ -94,7 +94,7 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore {

    @Override
    public List<PublishEvent> retrievePersistedPublishes(String clientID) {
-        return m_persistentMessageStore.get(clientID);
+        return new ArrayList<>(m_persistentMessageStore.get(clientID));
    }
    
    @Override
@@ -248,6 +248,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
        for (PublishEvent pubEvt : publishedEvents) {
            sendPublish(pubEvt.getClientID(), pubEvt.getTopic(), pubEvt.getQos(),
                   pubEvt.getMessage(), false, pubEvt.getMessageID());
+            m_messagesStore.cleanPersistedPublishMessage(clientID, pubEvt.getMessageID());
        }
    }
    
@@ -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);
    }

@@ -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<Subscription> 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() {
@@ -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