From aef6991eb40bb490e24758f103627cab68bb298d Mon Sep 17 00:00:00 2001 From: andrea Date: Sat, 18 Aug 2012 17:33:30 +0200 Subject: [PATCH] Modified client's subscribe to handle adhere to QoS 1 message type and handle retries --- .../org/dna/mqtt/moquette/client/Client.java | 115 ++++++++++++------ .../moquette/client/ClientMQTTHandler.java | 2 +- .../spi/impl/SubscriptionsStore.java | 3 +- 3 files changed, 82 insertions(+), 38 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 0a58c8fa..9527a909 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 @@ -1,10 +1,11 @@ package org.dna.mqtt.moquette.client; -import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.NetworkInterface; import java.util.HashMap; +import java.util.HashSet; import java.util.Map; +import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; @@ -20,6 +21,7 @@ import org.apache.mina.filter.codec.ProtocolCodecFilter; import org.apache.mina.filter.codec.demux.DemuxingProtocolDecoder; import org.apache.mina.filter.codec.demux.DemuxingProtocolEncoder; import org.apache.mina.transport.socket.nio.NioSocketConnector; +import org.dna.mqtt.commons.MessageIDGenerator; import org.dna.mqtt.moquette.ConnectionException; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.PublishException; @@ -50,11 +52,15 @@ import org.slf4j.LoggerFactory; * Pass to the constructor the host and port to connect, invoke the connect * message, invoke publish to send notifications messages, invoke subscribe with * a callback to be notified when something happen on the desired topic. + * + * NB this class is not thread safe so MUST be used by only one thread at time. * * @author andrea */ public final class Client { + final static int DEFAULT_RETRIES = 3; + final static int RETRIES_QOS_GT0 = 3; private static final Logger LOG = LoggerFactory.getLogger(Client.class); // private static final String HOSTNAME = /*"localhost"*/ "127.0.0.1"; // private static final int PORT = Server.PORT; @@ -62,7 +68,6 @@ public final class Client { 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; private String m_hostname; private int m_port; @@ -71,6 +76,7 @@ public final class Client { private IoSession m_session; private CountDownLatch m_connectBarrier; private CountDownLatch m_subscribeBarrier; + private int m_receivedSubAckMessageID; private byte m_returnCode; //TODO synchronize the access //Refact the da model should be a list of callback for each topic @@ -78,6 +84,9 @@ public final class Client { private ScheduledExecutorService m_scheduler; private ScheduledFuture m_pingerHandler; private String m_macAddress; + private MessageIDGenerator m_messageIDGenerator = new MessageIDGenerator(); + /**Maintain the list of in flight messages for QoS > 0*/ +// private Set m_inflightIDs = new HashSet(); private String m_clientID; @@ -248,43 +257,66 @@ public final class Client { SubscribeMessage msg = new SubscribeMessage(); msg.addSubscription(new SubscribeMessage.Couple((byte) AbstractMessage.QOSType.MOST_ONE.ordinal(), topic)); msg.setQos(AbstractMessage.QOSType.LEAST_ONE); - - WriteFuture wf = m_session.write(msg); - try { - wf.await(); - } catch (InterruptedException ex) { - LOG.error(null, ex); - throw new SubscribeException(ex); - } - LOG.info("subscribe message sent"); - - Throwable ex = wf.getException(); - if (ex != null) { - throw new SubscribeException(ex); - } - + int messageID = m_messageIDGenerator.next(); + msg.setMessageID(messageID); +// m_inflightIDs.add(messageID); register(topic, publishCallback); - - //wait for the SubAck - m_subscribeBarrier = new CountDownLatch(1); - - //suspend until the server respond with CONN_ACK - boolean unlocked = false; + + boolean unlocked = false; //true when a SUBACK is received try { - LOG.info("subscribe waiting for suback"); - unlocked = m_subscribeBarrier.await(SUBACK_TIMEOUT, TimeUnit.MILLISECONDS); //TODO parametrize - } catch (InterruptedException iex) { - throw new SubscribeException(iex); - } - - //if not arrive into certain limit, raise an error - if (!unlocked) { - throw new SubscribeException("Subscribe timeout elapsed unless server responded with a SUB_ACK"); + for (int retries = 0; retries < RETRIES_QOS_GT0 || !unlocked; retries++) { + if (retries > 0) { + msg.setDupFlag(true); + } + + WriteFuture wf = m_session.write(msg); + try { + wf.await(); + } catch (InterruptedException ex) { + LOG.error(null, ex); + throw new SubscribeException(ex); + } + LOG.info("subscribe message sent"); + + Throwable ex = wf.getException(); + if (ex != null) { + throw new SubscribeException(ex); + } + + //wait for the SubAck + m_subscribeBarrier = new CountDownLatch(1); + + //suspend until the server respond with CONN_ACK + try { + LOG.info("subscribe waiting for suback"); + unlocked = m_subscribeBarrier.await(SUBACK_TIMEOUT, TimeUnit.MILLISECONDS); //TODO parametrize + } catch (InterruptedException iex) { + throw new SubscribeException(iex); + } + } + + //if not arrive into certain limit, raise an error + if (!unlocked) { + throw new SubscribeException(String.format("Server doesn't replyed with a SUB_ACK after %d replies", RETRIES_QOS_GT0)); + } else { + //check if message ID match + if (m_receivedSubAckMessageID != messageID) { + throw new SubscribeException(String.format("Server replyed with " + + "a broken MessageID in SUB_ACK, expected %d but received %d", + messageID, m_receivedSubAckMessageID)); + } + } + } catch(Exception ex) { + //in case errors arise, remove the registration because the subscription + // hasn't get well + unregister(topic); + throw new MQTTException(ex); } +// register(topic, publishCallback); updatePinger(); - //TODO check the ACK messageID only for QoS 1 and 2 + //TODO check the ACK messageID } /** @@ -296,13 +328,20 @@ public final class Client { m_connectBarrier.countDown(); } - protected void subscribeAckCallback() { + protected void subscribeAckCallback(int messageID) { LOG.info("subscribeAckCallback invoked"); m_subscribeBarrier.countDown(); + m_receivedSubAckMessageID = messageID; } protected void publishCallback(String topic, byte[] payload) { - m_subscribersList.get(topic).published(topic, payload); + IPublishCallback callback = m_subscribersList.get(topic); + if (callback == null) { + String msg = String.format("Can't find any publish callback fr topic %s", topic); + LOG.error(msg); + throw new MQTTException(msg); + } + callback.published(topic, payload); } /** @@ -365,4 +404,10 @@ public final class Client { //register the publishCallback in some registry to be notified m_subscribersList.put(topic, publishCallback); } + + + //Remove the registration of the callback from the topic + private void unregister(String topic) { + m_subscribersList.remove(topic); + } } 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 57db0ab2..7fc689db 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 @@ -51,7 +51,7 @@ public class ClientMQTTHandler extends IoHandlerAdapter { } private void handleSubscribeAck(IoSession session, SubAckMessage subAckMessage) { - m_callback.subscribeAckCallback(); + m_callback.subscribeAckCallback(subAckMessage.getMessageID()); } private void handlePublish(IoSession session, PublishMessage pubMessage) { diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java index 481282c0..44f26c67 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java @@ -1,6 +1,5 @@ package org.dna.mqtt.moquette.messaging.spi.impl; -import java.io.IOException; import java.text.ParseException; import java.util.ArrayList; import java.util.List; @@ -16,7 +15,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * Represents a tree of published topics + * Represents a tree of topics subscriptions. * * @author andrea */