Re-added FuseSource MQTT client because simplify certain testing use cases

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

@@ -0,0 +1,108 @@
+package org.dna.mqtt.moquette.server;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import org.fusesource.mqtt.client.BlockingConnection;
+import org.fusesource.mqtt.client.MQTT;
+import org.fusesource.mqtt.client.QoS;
+import org.fusesource.mqtt.client.Topic;
+import org.fusesource.mqtt.client.Message;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import static org.junit.Assert.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ *
+ * @author andrea
+ */
+public class ServerIntegrationFuseTest {
+    private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class);
+
+    Server m_server;
+    MQTT m_mqtt;
+    
+//mqtt.setHost("test.mosquitto.org", 1883);
+
+
+    protected void startServer() throws IOException {
+        m_server = new Server();
+        m_server.startServer();
+    }
+
+    @Before
+    public void setUp() throws Exception {
+        startServer();
+
+        m_mqtt = new MQTT();
+        m_mqtt.setHost("localhost", 1883);
+    }
+
+    @After
+    public void tearDown() throws Exception {
+//        if (m_mqtt.isConnected()) {
+//            m_mqtt.disconnect();
+//        }
+
+        m_server.stopServer();
+        File dbFile = new File(Server.STORAGE_FILE_PATH);
+        if (dbFile.exists()) {
+            dbFile.delete();
+        }
+    }
+    
+    @Test
+    public void checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1() throws Exception {
+        LOG.info("*** checkReplayofStoredPublishResumeAfter_a_disconnect_cleanSessionFalseQoS1 ***");
+        MQTT mqtt = new MQTT();
+        mqtt.setHost("localhost", 1883); 
+        mqtt.setClientId("Publisher");
+        BlockingConnection publisher = mqtt.blockingConnection();
+        publisher.connect();
+        
+        m_mqtt.setHost("localhost", 1883); 
+        m_mqtt.setCleanSession(false);
+        m_mqtt.setClientId("Subscriber");
+        BlockingConnection subscriber = m_mqtt.blockingConnection();
+        subscriber.connect();
+        Topic[] topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)};
+        byte[] qoses = subscriber.subscribe(topics);
+        
+        //force the publisher to send
+        publisher.publish("/topic", "Hello world MQTT!!-1".getBytes(), QoS.AT_LEAST_ONCE, false);
+        
+        //read the first message and drop the connection
+        Message msg = subscriber.receive();
+        msg.ack();
+        assertEquals("Hello world MQTT!!-1", new String(msg.getPayload()));
+        subscriber.disconnect();
+        
+        publisher.publish("/topic", "Hello world MQTT!!-2".getBytes(), QoS.AT_LEAST_ONCE, false);
+        publisher.publish("/topic", "Hello world MQTT!!-3".getBytes(), QoS.AT_LEAST_ONCE, false);
+        
+        //reconnect and expect to receive the hello 2 message
+        subscriber = m_mqtt.blockingConnection();
+        subscriber.connect();
+        topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)};
+        qoses = subscriber.subscribe(topics);
+        msg = subscriber.receive();
+        msg.ack();
+        assertEquals("Hello world MQTT!!-2", new String(msg.getPayload()));
+        subscriber.disconnect();
+        
+        subscriber = m_mqtt.blockingConnection();
+        subscriber.connect();
+        topics = new Topic[]{new Topic("/topic", QoS.AT_LEAST_ONCE)};
+        qoses = subscriber.subscribe(topics);
+        msg = subscriber.receive();
+        assertEquals("Hello world MQTT!!-3", new String(msg.getPayload()));
+        msg.ack();
+        //TODO check topic and content
+        subscriber.disconnect();
+    }
+}

broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java

@@ -334,7 +334,7 @@ public class ServerIntegrationPahoTest {
         m_client.subscribe("/topic", 2);
         m_client.disconnect();
 
-        //publish a QoS 1 message another client publish a message on the topic
+        //publish a QoS 2 message another client publish a message on the topic
         publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 2);
         m_callback.reinit();
         m_client.connect(options);
@@ -343,7 +343,34 @@ public class ServerIntegrationPahoTest {
         assertNotNull(m_callback.getMessage());
         assertEquals("Hello MQTT", m_callback.getMessage().toString());
     }
-
+    
+    @Test
+    public void avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1() throws Exception {
+        LOG.info("*** avoidMultipleNotificationsAfterMultipleReconnection_cleanSessionFalseQoS1, issue #16 ***");
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setCleanSession(false);
+        m_client.connect(options);
+        m_client.subscribe("/topic", 1);
+        m_client.disconnect();
+        
+        publishFromAnotherClient("/topic", "Hello MQTT 1".getBytes(), 1);
+        m_callback.reinit();
+        m_client.connect(options);
+        
+        assertNotNull(m_callback);
+        assertNotNull(m_callback.getMessage());
+        assertEquals("Hello MQTT 1", m_callback.getMessage().toString());
+        m_client.disconnect();
+        
+        //publish other message
+        publishFromAnotherClient("/topic", "Hello MQTT 2".getBytes(), 1);
+        
+        //reconnect the second time
+        m_callback.reinit();
+        m_client.connect(options);
+        assertNotNull(m_callback);
+        assertNotNull(m_callback.getMessage());
+        assertEquals("Hello MQTT 2", m_callback.getMessage().toString());
+    }
 
 }
-