Fixed issue #15, missed notification of publish due to bad re-activation on a nested topic tree.

ChangeLog.txt

@@ -8,3 +8,4 @@ Version 0.3:
 
 Version 0.4:
     Dropped support for MINA
+    Fix Issue 15 (bad re-activation of client's subscriptions after a recconection)

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java

@@ -148,7 +148,7 @@ class TreeNode {
 
         //go deep
         for (TreeNode child : m_children) {
-            child.deactivate(clientID);
+            child.activate(clientID);
         }
 
     }

broker/src/main/resources/log4j.properties

@@ -12,10 +12,15 @@ log4j.rootLogger=ERROR, stdout, file
 log4j.logger.org.dna.mqtt=INFO
 log4j.logger.org.dna.mqtt.moquette.server.ServerIntegrationFuseTest=DEBUG
 log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.ProtocolProcessor=INFO
+log4j.logger.org.dna.mqtt.moquette.messaging.spi.impl.SimpleMessaging=WARN
+#Protocol parsing
+log4j.logger.org.dna.mqtt.moquette.server.netty.NettyMQTTHandler=WARN
+log4j.logger.org.dna.mqtt.moquette.parser.netty=WARN
 
 # stdout appender is set to be a ConsoleAppender.
 log4j.appender.stdout=org.apache.log4j.ConsoleAppender
-log4j.appender.stdout.Threshold=WARN
+#log4j.appender.stdout.Threshold=WARN
+log4j.appender.stdout.Threshold=INFO
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n
 

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

@@ -62,6 +62,8 @@ public class ServerIntegrationPahoTest {
     private static final Logger LOG = LoggerFactory.getLogger(ServerIntegrationPahoTest.class);
 
     static MqttClientPersistence s_dataStore;
+    static MqttClientPersistence s_pubDataStore;
+    
 
     Server m_server;
     IMqttClient m_client;
@@ -71,6 +73,7 @@ public class ServerIntegrationPahoTest {
     public static void beforeTests() {
         String tmpDir = System.getProperty("java.io.tmpdir");
         s_dataStore = new MqttDefaultFilePersistence(tmpDir);
+        s_pubDataStore = new MqttDefaultFilePersistence(tmpDir + File.separator + "publisher");
     }
 
     protected void startServer() throws IOException {
@@ -275,9 +278,29 @@ public class ServerIntegrationPahoTest {
 
         assertEquals("Hello MQTT", m_callback.getMessage().toString());
     }
-
+    
+    @Test
+    public void checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession() throws Exception {
+        LOG.info("*** checkReceivePublishedMessage_after_a_reconnect_with_notCleanSession ***");
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setCleanSession(false);
+        m_client.connect(options);
+        m_client.subscribe("/topic", 1);
+        m_client.disconnect();
+        
+        m_client.connect(options);
+        m_client.subscribe("/topic", 1);
+        
+        //publish a QoS 1 message another client publish a message on the topic
+        publishFromAnotherClient("/topic", "Hello MQTT".getBytes(), 1); 
+        
+        //Verify that after a reconnection the client receive the message
+        assertNotNull(m_callback.getMessage());
+        assertEquals("Hello MQTT", m_callback.getMessage().toString());
+    }
+ 
     private void publishFromAnotherClient(String topic, byte[] payload, int qos) throws Exception {
-        IMqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClientPUB", s_dataStore);
+        IMqttClient anotherClient = new MqttClient("tcp://localhost:1883", "TestClientPUB", s_pubDataStore);
         anotherClient.connect();
         anotherClient.publish(topic, payload, qos, false);
         anotherClient.disconnect();

tools_scripts/issue15_receiver.groovy

@@ -0,0 +1,48 @@
+@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5')
+
+import java.net.URISyntaxException
+
+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
+
+class Reciever {
+    MQTT mqtt = new MQTT()
+    BlockingConnection connection
+    
+    public void connectAndSubscribe(boolean cleanSession) {
+        mqtt.setCleanSession(cleanSession)
+        mqtt.setHost("localhost", 1883)
+
+        mqtt.setClientId("reciever")
+        connection = mqtt.blockingConnection()
+        connection.connect()
+        Topic[] topics = [new Topic("/topic", QoS.AT_LEAST_ONCE)]
+        byte[] qoses = connection.subscribe(topics)
+    }
+}
+
+
+Reciever r = new Reciever()
+r.connectAndSubscribe(false)
+Message message
+3.times { i ->
+    message = r.connection.receive()
+    println "Topic: ${message.topic}, payload: ${new String(message.payload)}"
+    message.ack()
+}
+r.connection.kill()
+//r.connection.disconnect()
+
+Thread.sleep(3000)
+
+r.connectAndSubscribe(false)
+3.times { i ->
+    message = r.connection.receive()
+    println "Topic: ${message.topic}, payload: ${new String(message.payload)}"
+    message.ack()
+}
+r.connection.disconnect()
+

tools_scripts/issue15_sender.groovy

@@ -0,0 +1,27 @@
+//@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/')
+@Grab(group='org.fusesource.mqtt-client', module='mqtt-client', version='1.5')
+
+import java.net.URISyntaxException
+
+import org.fusesource.mqtt.client.BlockingConnection
+import org.fusesource.mqtt.client.MQTT
+import org.fusesource.mqtt.client.QoS
+
+
+MQTT mqtt = new MQTT()
+//mqtt.setHost("test.mosquitto.org", 1883);
+mqtt.setHost("localhost", 1883)
+
+BlockingConnection connection = mqtt.blockingConnection()
+connection.connect()
+
+//for (int i = 0; i < 1000; i++) {
+(1..1000).each{ i -> 
+//        LOG.info("Publishing");
+        String payload = "Hello world MQTT!!${i}"
+        connection.publish("/topic", payload.getBytes(), QoS.AT_LEAST_ONCE, false)
+        println payload
+        Thread.sleep 1000
+}
+
+connection.disconnect()