Fixed error in storage of QoS publish messages, before stored with current clientID and not matching subscription's clientID

parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java

@@ -34,8 +34,7 @@ public class HawtDBStorageService implements IStorageService {
     //bind clientID+MsgID -> evt message published
     private SortedIndex<String, PublishEvent> m_inflightStore;
 
-    //persistent Map of clientID, list of Subscriptions
+    //persistent Map of clientID, set of Subscriptions
-    //TODO move to Set to avoid double subscriptions
     private SortedIndex<String, Set<Subscription>> m_persistentSubscriptions;
 
     public HawtDBStorageService() {

parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java

@@ -125,6 +125,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
 
     public void republishStored(String clientID) {
         //create the event to push
+        LOG.debug("republishStored invoked to publish soterd messages for clientID " + clientID);
         disruptorPublish(new RepublishEvent(clientID));
     }
 
@@ -176,8 +177,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     }
     
     public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
-        LOG.debug("onEvent processing event");
         MessagingEvent evt = t.getEvent();
+        LOG.debug("onEvent processing messaging event " + evt);
         if (evt instanceof PublishEvent) {
             processPublish((PublishEvent) evt);
         } else if (evt instanceof StopEvent) {
@@ -307,6 +308,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
             }
         }
 
+        subscriptions.connect(msg.getClientID());
+
         //handle clean session flag
         if (msg.isCleanSession()) {
             //remove all prev subscriptions
@@ -323,7 +326,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     }
 
     protected void processPublish(PublishEvent evt) {
-        LOG.debug("processPublish invoked");
+        LOG.debug("processPublish invoked with " + evt);
         final String topic = evt.getTopic();
         final QOSType qos = evt.getQos();
         final byte[] message = evt.getMessage();
@@ -346,9 +349,12 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
                 //QoS 1 or 2
                 //if the target subscription is not clean session and is not connected => store it
                 if (!sub.isCleanSession() && !sub.isActive()) {
-                    m_storageService.storePublishForFuture(evt);
+                    //clone the event with matching clientID
+                    PublishEvent newPublishEvt = new PublishEvent(topic, qos, message, retain, sub.getClientId(), evt.getMessageID(), null);
+                    m_storageService.storePublishForFuture(newPublishEvt);
+                } else  {
+                    notify(new NotifyEvent(sub.clientId, topic, qos, message, false));
                 }
-                notify(new NotifyEvent(sub.clientId, topic, qos, message, false));
             }
         }
 
@@ -396,8 +402,9 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         //TODO by now it handles only QoS 0 messages
         for (int i = 0; i < msg.subscriptions().size(); i++) {
             ackMessage.addType(QOSType.MOST_ONE);
+            ackMessage.addType(QOSType.LEAST_ONE);
         }
-        LOG.info("replying with SubAct to MSG ID {0}", msg.getMessageID());
+        LOG.info("replying with SubAct to MSG ID " + msg.getMessageID());
         session.write(ackMessage);
 
     }
@@ -447,8 +454,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     }
 
     private void processRepublish(RepublishEvent evt) throws InterruptedException {
+        LOG.debug("processRepublish invoked");
         List<PublishEvent> publishedEvents = m_storageService.retrivePersistedPublishes(evt.getClientID());
         if (publishedEvents == null) {
+            LOG.debug("processRepublish, no stored publish events");
             return;
         }
 
@@ -470,7 +479,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
             pubMessage.setMessageID(evt.getMessageID());
         }
 
-        LOG.debug("notify invoked");
         try {
             assert m_clientIDs != null;
             LOG.debug("clientIDs are " + m_clientIDs);

parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationFuseTest.java

@@ -0,0 +1,135 @@
+package org.dna.mqtt.moquette.server;
+
+import org.dna.mqtt.moquette.client.Client;
+import org.dna.mqtt.moquette.client.IPublishCallback;
+import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
+import org.fusesource.mqtt.client.*;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+
+import static org.junit.Assert.*;
+
+
+/**
+ * Test the broker using the widely used FuseSource libraries
+ * */
+public class ServerIntegrationFuseTest {
+
+    boolean received;
+    Server server;
+
+    protected void startServer() throws IOException {
+        server = new Server();
+        server.startServer();
+    }
+
+    @Before
+    public void setUp() throws IOException {
+        startServer();
+    }
+
+    @After
+    public void tearDown() {
+        server.stopServer();
+        File dbFile = new File(Server.STORAGE_FILE_PATH);
+        if (dbFile.exists()) {
+            dbFile.delete();
+        }
+    }
+
+    @Ignore
+    public void testSubscribe() throws Exception {
+//        startServer();
+        MQTT mqtt = new MQTT();
+        mqtt.setHost("localhost", Server.PORT);
+        mqtt.setClientId("TestClient");
+        FutureConnection connection = mqtt.futureConnection();
+        connection.connect().await();
+
+        Topic[] topics = {new Topic("/topic", QoS.AT_MOST_ONCE)};
+        Future<byte[]> futSub = connection.subscribe(topics);
+
+        connection.publish("/topic", "Test my payload".getBytes(), QoS.AT_MOST_ONCE, false).await();
+
+        byte[] qoses = futSub.await();
+        assertEquals(1, qoses.length);
+        connection.disconnect().await();
+    }
+
+    @Ignore
+    public void testPublishWithQoS1() throws Exception {
+        MQTT mqtt = new MQTT();
+        mqtt.setHost("localhost", Server.PORT);
+        mqtt.setClientId("TestClient");
+        FutureConnection connection = mqtt.futureConnection();
+        connection.connect().await();
+
+        //subscribe to a QoS 1 topic
+        Topic[] topics = {new Topic("/topic", QoS.AT_LEAST_ONCE)};
+        Future<byte[]> subscribeFuture = connection.subscribe(topics);
+        byte[] qoses = subscribeFuture.await();
+
+        Future<Message> receive = connection.receive();
+
+        //publish a QoS 1 message
+        Future<Void> f3 = connection.publish("/topic", "Hello MQTT".getBytes(), QoS.AT_LEAST_ONCE, false);
+
+        //verify the reception
+        Message message = receive.await();
+        message.ack();
+
+        assertEquals("Hello MQTT", new String(message.getPayload()));
+
+        connection.disconnect().await();
+    }
+
+    @Test
+    public void testPublishWithQoS1_notCleanSession() throws Exception {
+        MQTT mqtt = new MQTT();
+        mqtt.setHost("localhost", Server.PORT);
+        mqtt.setClientId("TestClient");
+        mqtt.setCleanSession(false);
+        FutureConnection connection = mqtt.futureConnection() ;
+        connection.connect().await();
+
+        //subscribe to a QoS 1 topic
+        Topic[] topics = {new Topic("/topic", QoS.AT_LEAST_ONCE)};
+        Future<byte[]> subscribeFuture = connection.subscribe(topics);
+        byte[] qoses = subscribeFuture.await();
+
+        connection.disconnect().await();
+
+        //publish a QoS 1 message another client publish a message on the topic
+        publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), QoS.AT_LEAST_ONCE);
+
+
+        //Reconnect
+        connection = mqtt.futureConnection();
+        connection.connect().await();
+        Future<Message> receive = connection.receive();
+
+        //verify the reception
+        Message message = receive.await();
+        message.ack();
+
+        assertEquals("Hello MQTT", new String(message.getPayload()));
+
+        connection.disconnect().await();
+    }
+
+    private void publishFromAnotherClient(String topic, byte[] payload, QoS qos) throws Exception {
+        MQTT mqttPub = new MQTT();
+        mqttPub.setHost("localhost", Server.PORT);
+        mqttPub.setClientId("TestClientPUB");
+        FutureConnection connectionPub = mqttPub.futureConnection();
+        connectionPub.connect().await();
+        Future<Void> futurePublish = connectionPub.publish(topic, payload, qos, false);
+        futurePublish.await();
+        connectionPub.disconnect().await();
+    }
+}