From b32bbe46293eff4419910d89279d153ce841b36f Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Fri, 27 Jul 2012 00:17:43 +0200 Subject: [PATCH] Added skeleton for client side of subscription --- .../dna/mqtt/moquette/SubscribeException.java | 16 +++++ .../org/dna/mqtt/moquette/client/Client.java | 61 ++++++++++++++++++- .../moquette/client/ClientMQTTHandler.java | 8 +++ .../moquette/client/IPublishCallback.java | 10 +++ .../proto/messages/SubscribeMessage.java | 5 ++ parser/src/main/resources/log4j.properties | 9 +++ .../server/ServerIntegrationTest.java | 8 ++- 7 files changed, 114 insertions(+), 3 deletions(-) create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/SubscribeException.java create mode 100644 parser/src/main/java/org/dna/mqtt/moquette/client/IPublishCallback.java create mode 100644 parser/src/main/resources/log4j.properties diff --git a/parser/src/main/java/org/dna/mqtt/moquette/SubscribeException.java b/parser/src/main/java/org/dna/mqtt/moquette/SubscribeException.java new file mode 100644 index 00000000..54443a2e --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/SubscribeException.java @@ -0,0 +1,16 @@ +package org.dna.mqtt.moquette; + +/** + * + * @author andrea + */ +public class SubscribeException extends ConnectionException { + + public SubscribeException(String msg) { + super(msg); + } + + public SubscribeException(Throwable e) { + super(e); + } +} 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 f412cbe7..2f71d29d 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 @@ -19,15 +19,19 @@ import org.apache.mina.transport.socket.nio.NioSocketConnector; import org.dna.mqtt.moquette.ConnectionException; import org.dna.mqtt.moquette.MQTTException; import org.dna.mqtt.moquette.PublishException; +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.PublishEncoder; +import org.dna.mqtt.moquette.proto.SubAckDecoder; +import org.dna.mqtt.moquette.proto.SubscribeEncoder; 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.PublishMessage; +import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; import org.dna.mqtt.moquette.server.Server; /** @@ -44,7 +48,8 @@ public final class Client { private static final Logger LOG = Logger.getLogger(Client.class.getName()); // private static final String HOSTNAME = /*"localhost"*/ "127.0.0.1"; // private static final int PORT = Server.PORT; - private static final long CONNECT_TIMEOUT = 3 * 1000L; // 30 seconds + private static final long CONNECT_TIMEOUT = 3 * 1000L; // 3 seconds + private static final long SUBACK_TIMEOUT = 4 * 1000L; final static int DEFAULT_RETRIES = 3; private int m_connectRetries = DEFAULT_RETRIES; @@ -55,6 +60,7 @@ public final class Client { private IoConnector m_connector; private IoSession m_session; private CountDownLatch m_connectBarrier; + private CountDownLatch m_subscribeBarrier; private byte m_returnCode; public Client(String host, int port) { @@ -66,15 +72,17 @@ public final class Client { protected void init() { DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); decoder.addMessageDecoder(new ConnAckDecoder()); + decoder.addMessageDecoder(new SubAckDecoder()); DemuxingProtocolEncoder encoder = new DemuxingProtocolEncoder(); encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); + encoder.addMessageEncoder(SubscribeMessage.class, new SubscribeEncoder()); encoder.addMessageEncoder(DisconnectMessage.class, new DisconnectEncoder()); m_connector = new NioSocketConnector(); - m_connector.getFilterChain().addLast("logger", new LoggingFilter()); +// m_connector.getFilterChain().addLast("logger", new LoggingFilter()); m_connector.getFilterChain().addLast("codec", new ProtocolCodecFilter(encoder, decoder)); m_connector.setHandler(new ClientMQTTHandler(this)); @@ -180,6 +188,48 @@ public final class Client { } } + public void subscribe(String topic, IPublishCallback publishCallback) { + LOG.info("subscribe invoked"); + 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.log(Level.SEVERE, null, ex); + throw new SubscribeException(ex); + } + LOG.info("subscribe message sent"); + + Throwable ex = wf.getException(); + if (ex != null) { + throw new SubscribeException(ex); + } + + //TODO register the publishCallback in some registry to be notified + + //wait for the SubAck + m_subscribeBarrier = new CountDownLatch(1); + + //suspend until the server respond with CONN_ACK + boolean unlocked = false; + 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"); + } + + //TODO check the ACK messageID + } + /** * TODO extract this SPI method in a SPI @@ -190,6 +240,13 @@ public final class Client { m_connectBarrier.countDown(); } + + protected void subscribeAckCallback() { + LOG.info("subscribeAckCallback invoked"); + m_subscribeBarrier.countDown(); + } + + public void close() { //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 b84648d9..460ea3df 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 @@ -6,6 +6,7 @@ import org.apache.mina.core.session.IoSession; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; +import org.dna.mqtt.moquette.proto.messages.SubAckMessage; /** * @@ -29,6 +30,9 @@ public class ClientMQTTHandler extends IoHandlerAdapter { case CONNACK: handleConnectAck(session, (ConnAckMessage) msg); break; + case SUBACK: + handleSubscribeAck(session, (SubAckMessage) msg); + break; // case SUBSCRIBE: // handleSubscribe(session, (SubscribeMessage) msg); // break; @@ -41,4 +45,8 @@ public class ClientMQTTHandler extends IoHandlerAdapter { private void handleConnectAck(IoSession session, ConnAckMessage connAckMessage) { m_callback.connectionAckCallback(connAckMessage.getReturnCode()); } + + private void handleSubscribeAck(IoSession session, SubAckMessage subAckMessage) { + m_callback.subscribeAckCallback(); + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/client/IPublishCallback.java b/parser/src/main/java/org/dna/mqtt/moquette/client/IPublishCallback.java new file mode 100644 index 00000000..0fdede1b --- /dev/null +++ b/parser/src/main/java/org/dna/mqtt/moquette/client/IPublishCallback.java @@ -0,0 +1,10 @@ +package org.dna.mqtt.moquette.client; + +/** + * + * @author andrea + */ +public interface IPublishCallback { + + void published(String topic, byte[] message); +} diff --git a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java b/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java index 118a9800..d93a52eb 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/proto/messages/SubscribeMessage.java @@ -29,6 +29,11 @@ public class SubscribeMessage extends MessageIDMessage { } private List m_subscriptions = new ArrayList(); + public SubscribeMessage() { + //Subscribe has always QoS 1 + m_qos = AbstractMessage.QOSType.LEAST_ONE; + } + public List subscriptions() { return m_subscriptions; } diff --git a/parser/src/main/resources/log4j.properties b/parser/src/main/resources/log4j.properties new file mode 100644 index 00000000..68b844a6 --- /dev/null +++ b/parser/src/main/resources/log4j.properties @@ -0,0 +1,9 @@ +# Set root logger level to DEBUG and its only appender to A1. +log4j.rootLogger=DEBUG, A1 + +# A1 is set to be a ConsoleAppender. +log4j.appender.A1=org.apache.log4j.ConsoleAppender + +# A1 uses PatternLayout. +log4j.appender.A1.layout=org.apache.log4j.PatternLayout +log4j.appender.A1.layout.ConversionPattern=%-4r [%t] %-5p %c{1} %x - %m%n \ No newline at end of file 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 6bf26994..1a02890e 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 @@ -3,6 +3,7 @@ package org.dna.mqtt.moquette.server; import java.io.IOException; import org.apache.mina.core.service.IoConnector; import org.dna.mqtt.moquette.client.Client; +import org.dna.mqtt.moquette.client.IPublishCallback; import org.junit.Test; /** @@ -24,7 +25,12 @@ public class ServerIntegrationTest { client.connect(); client.publish("/topic", "Test my payload".getBytes()); - //TODO client.subscribe("/topic", callback) + client.subscribe("/topic", new IPublishCallback() { + + public void published(String topic, byte[] message) { + throw new UnsupportedOperationException("Not supported yet."); + } + }); client.close(); }