Fixed timer for Ping requests

parser/src/main/java/org/dna/mqtt/moquette/client/Client.java

@@ -26,6 +26,7 @@ import org.dna.mqtt.moquette.proto.ConnAckDecoder;
 import org.dna.mqtt.moquette.proto.ConnectEncoder;
 import org.dna.mqtt.moquette.proto.DisconnectEncoder;
 import org.dna.mqtt.moquette.proto.PingReqEncoder;
+import org.dna.mqtt.moquette.proto.PingRespDecoder;
 import org.dna.mqtt.moquette.proto.PublishDecoder;
 import org.dna.mqtt.moquette.proto.PublishEncoder;
 import org.dna.mqtt.moquette.proto.SubAckDecoder;
@@ -77,7 +78,13 @@ public final class Client {
     private Map<String, IPublishCallback> m_subscribersList = new HashMap<String, IPublishCallback>();
     private ScheduledExecutorService m_scheduler;
     private ScheduledFuture m_pingerHandler;
-    private boolean m_executedOperation;
+    final Runnable pingerDeamon = new Runnable() {
+            public void run() { 
+                LOG.debug("Pingreq sent");
+                //send a ping req
+                m_session.write(new PingReqMessage());
+            }
+        };
 
     public Client(String host, int port) {
         m_hostname = host;
@@ -90,6 +97,7 @@ public final class Client {
         decoder.addMessageDecoder(new ConnAckDecoder());
         decoder.addMessageDecoder(new SubAckDecoder());
         decoder.addMessageDecoder(new PublishDecoder());
+        decoder.addMessageDecoder(new PingRespDecoder());
 
         DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder();
         encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder());
@@ -172,18 +180,7 @@ public final class Client {
             throw new ConnectionException(errMsg);
         }
         
-        m_executedOperation = true;
-        final Runnable pingerDeamon = new Runnable() {
-            public void run() { 
-                if (!m_executedOperation) {
-                    //send a ping req
-                    PingReqMessage msg = new PingReqMessage();
-                    m_session.write(msg);
-                } 
-                m_executedOperation = false;
-            }
-        };
-        m_pingerHandler = m_scheduler.scheduleWithFixedDelay(pingerDeamon, KEEPALIVE_SECS, KEEPALIVE_SECS, TimeUnit.SECONDS);
+        updatePinger();
     }
     
     /**
@@ -211,7 +208,7 @@ public final class Client {
             throw new PublishException(ex);
         }
         
-        m_executedOperation = true;
+        updatePinger();
     }
     
     public void subscribe(String topic, IPublishCallback publishCallback) {
@@ -254,7 +251,7 @@ public final class Client {
             throw new SubscribeException("Subscribe timeout elapsed unless server responded with a SUB_ACK");
         }
         
-        m_executedOperation = true;
+        updatePinger();
         
         //TODO check the ACK messageID
     }
@@ -281,6 +278,18 @@ public final class Client {
     }
     
     
+    /**
+     * In the current pinger is not ye executed, then cancel it and schedule 
+     * another by KEEPALIVE_SECS
+     */
+    private void updatePinger() {
+        if (m_pingerHandler != null) {
+            m_pingerHandler.cancel(false);
+        }
+        m_pingerHandler = m_scheduler.scheduleWithFixedDelay(pingerDeamon, KEEPALIVE_SECS, KEEPALIVE_SECS, TimeUnit.SECONDS);
+    }
+    
+    
     public void close() {
         //stop the pinger
         m_pingerHandler.cancel(false);

parser/src/main/java/org/dna/mqtt/moquette/client/Main.java

@@ -1,7 +1,6 @@
 package org.dna.mqtt.moquette.client;
 
 import java.io.IOException;
-import org.dna.mqtt.moquette.server.Server;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -14,24 +13,22 @@ public class Main {
     
     static boolean received;
     
-    public static void main(String[] args) throws IOException {
-        Client client = new Client("localhost", Server.PORT);
+    public static void main(String[] args) throws IOException, InterruptedException {
+        Client client = new Client("test.mosquitto.org", 1883);
         client.connect();
         
-        
         client.subscribe("/topic", new IPublishCallback() {
 
             public void published(String topic, byte[] message) {
-                LOG.info("Received publush notification");
                 received = true;
             }
         });
         
+        Thread.sleep(5000);
+        
         client.publish("/topic", "Test my payload".getBytes());
+        
         client.close();
-        
-        
-        assert received;
     }
     
 }

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

@@ -37,6 +37,7 @@ public class ServerIntegrationTest {
     public void testSubscribe() throws IOException, InterruptedException {
 //        startServer();
         Client client = new Client("localhost", Server.PORT);
+//        Client client = new Client("test.mosquitto.org", 1883);
         client.connect();
         
         
@@ -49,10 +50,8 @@ public class ServerIntegrationTest {
         
         client.publish("/topic", "Test my payload".getBytes());
         
-        
         client.close();
         
-        
         assertTrue(received);
     }