From 343f36c4f5194e5d2ed8ba9ce7a1a50d87441886 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Wed, 8 Aug 2012 23:31:26 +0200 Subject: [PATCH] Added cyclic ping request thread. --- .../org/dna/mqtt/moquette/client/Client.java | 37 ++++++++++++++++++- .../moquette/client/ClientMQTTHandler.java | 2 + 2 files changed, 37 insertions(+), 2 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 fa6bc0c8..f52f7a24 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 @@ -4,6 +4,9 @@ import java.net.InetSocketAddress; import java.util.HashMap; import java.util.Map; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import org.apache.mina.core.RuntimeIoException; import org.apache.mina.core.future.ConnectFuture; @@ -22,6 +25,7 @@ import org.dna.mqtt.moquette.SubscribeException; 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.PublishDecoder; import org.dna.mqtt.moquette.proto.PublishEncoder; import org.dna.mqtt.moquette.proto.SubAckDecoder; @@ -30,6 +34,7 @@ import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; import org.dna.mqtt.moquette.proto.messages.DisconnectMessage; +import org.dna.mqtt.moquette.proto.messages.PingReqMessage; import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; import org.dna.mqtt.moquette.server.Server; @@ -52,6 +57,8 @@ public final class Client { // private static final int PORT = Server.PORT; private static final long CONNECT_TIMEOUT = 3 * 1000L; // 3 seconds private static final long SUBACK_TIMEOUT = 4 * 1000L; + private static final int KEEPALIVE_SECS = 3; + private static final int NUM_SCHEDULER_TIMER_THREAD = 1; final static int DEFAULT_RETRIES = 3; private int m_connectRetries = DEFAULT_RETRIES; @@ -68,6 +75,9 @@ public final class Client { //TODO synchronize the access //Refact the da model should be a list of callback for each topic private Map m_subscribersList = new HashMap(); + private ScheduledExecutorService m_scheduler; + private ScheduledFuture m_pingerHandler; + private boolean m_executedOperation; public Client(String host, int port) { m_hostname = host; @@ -86,7 +96,8 @@ public final class Client { encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); encoder.addMessageEncoder(SubscribeMessage.class, new SubscribeEncoder()); encoder.addMessageEncoder(DisconnectMessage.class, new DisconnectEncoder()); - + encoder.addMessageEncoder(PingReqMessage.class, new PingReqEncoder()); + m_connector = new NioSocketConnector(); // m_connector.getFilterChain().addLast("logger", new LoggingFilter()); @@ -95,6 +106,8 @@ public final class Client { m_connector.setHandler(new ClientMQTTHandler(this)); m_connector.getSessionConfig().setReadBufferSize(2048); m_connector.getSessionConfig().setIdleTime(IdleStatus.BOTH_IDLE, Server.DEFAULT_CONNECT_TIMEOUT); + + m_scheduler = Executors.newScheduledThreadPool(NUM_SCHEDULER_TIMER_THREAD); } public void connect() throws MQTTException { @@ -117,7 +130,7 @@ public final class Client { //send a message over the session ConnectMessage connMsg = new ConnectMessage(); - connMsg.setKeepAlive(3); + connMsg.setKeepAlive(KEEPALIVE_SECS); connMsg.setClientID("FAKE_MSG_ID1"); //TODO create a generator for the ID m_session.write(connMsg); @@ -158,6 +171,19 @@ 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); } /** @@ -184,6 +210,8 @@ public final class Client { if (ex != null) { throw new PublishException(ex); } + + m_executedOperation = true; } public void subscribe(String topic, IPublishCallback publishCallback) { @@ -226,6 +254,8 @@ public final class Client { throw new SubscribeException("Subscribe timeout elapsed unless server responded with a SUB_ACK"); } + m_executedOperation = true; + //TODO check the ACK messageID } @@ -252,6 +282,9 @@ public final class Client { public void close() { + //stop the pinger + m_pingerHandler.cancel(false); + //send the CLOSE message m_session.write(new DisconnectMessage()); diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java index a44094d9..57db0ab2 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/ClientMQTTHandler.java @@ -41,6 +41,8 @@ public class ClientMQTTHandler extends IoHandlerAdapter { case PUBLISH: handlePublish(session, (PublishMessage) msg); break; + case PINGRESP: + break; } }