From 937a7b706021ce63210aa9059a28630e2b8eb735 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Thu, 9 Aug 2012 22:40:23 +0200 Subject: [PATCH] Fixed timer for Ping requests --- .../org/dna/mqtt/moquette/client/Client.java | 39 ++++++++++++------- .../org/dna/mqtt/moquette/client/Main.java | 13 +++---- .../server/ServerIntegrationTest.java | 3 +- 3 files changed, 30 insertions(+), 25 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java b/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java index f52f7a24..5c6467e5 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/Client.java +++ b/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 m_subscribersList = new HashMap(); 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); diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/Main.java b/parser/src/main/java/org/dna/mqtt/moquette/client/Main.java index 1b2836d7..e224884b 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/Main.java +++ b/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; } } diff --git a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java b/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java index 7626076d..d5359d4e 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationTest.java +++ b/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); }