From d0677f1c44fbb5a93bfb90fa66e999cfa5ef63a6 Mon Sep 17 00:00:00 2001 From: andrea Date: Sun, 19 Aug 2012 12:25:02 +0200 Subject: [PATCH] Added unsubscribe handling to client lib --- .../org/dna/mqtt/moquette/client/Client.java | 43 ++++++++++++++++++- .../moquette/client/ClientMQTTHandler.java | 8 ++++ 2 files changed, 49 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 131d6a4e..8dc06cf6 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 @@ -33,6 +33,8 @@ import org.dna.mqtt.moquette.proto.PublishDecoder; 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.UnsubAckDecoder; +import org.dna.mqtt.moquette.proto.UnsubscribeEncoder; import org.dna.mqtt.moquette.proto.messages.AbstractMessage; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.ConnectMessage; @@ -41,6 +43,7 @@ import org.dna.mqtt.moquette.proto.messages.MessageIDMessage; 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.proto.messages.UnsubscribeMessage; import org.dna.mqtt.moquette.server.Server; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -115,6 +118,7 @@ public final class Client { DemuxingProtocolDecoder decoder = new DemuxingProtocolDecoder(); decoder.addMessageDecoder(new ConnAckDecoder()); decoder.addMessageDecoder(new SubAckDecoder()); + decoder.addMessageDecoder(new UnsubAckDecoder()); decoder.addMessageDecoder(new PublishDecoder()); decoder.addMessageDecoder(new PingRespDecoder()); @@ -122,6 +126,7 @@ public final class Client { encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); encoder.addMessageEncoder(SubscribeMessage.class, new SubscribeEncoder()); + encoder.addMessageEncoder(UnsubscribeMessage.class, new UnsubscribeEncoder()); encoder.addMessageEncoder(DisconnectMessage.class, new DisconnectEncoder()); encoder.addMessageEncoder(PingReqMessage.class, new PingReqEncoder()); @@ -261,7 +266,7 @@ public final class Client { // m_inflightIDs.add(messageID); register(topic, publishCallback); - boolean unlocked = false; //true when a SUBACK is received +// boolean unlocked = false; //true when a SUBACK is received try { manageSendQoS1(msg); // for (int retries = 0; retries < RETRIES_QOS_GT0 || !unlocked; retries++) { @@ -315,8 +320,35 @@ public final class Client { // register(topic, publishCallback); updatePinger(); + } + + + public void unsubscribe(String... topics) { + LOG.info("unsubscribe invoked"); + UnsubscribeMessage msg = new UnsubscribeMessage(); + for (String topic : topics) { + msg.addTopic(topic); + } + msg.setQos(AbstractMessage.QOSType.LEAST_ONE); + int messageID = m_messageIDGenerator.next(); + msg.setMessageID(messageID); +// m_inflightIDs.add(messageID); +// register(topic, publishCallback); + try { + manageSendQoS1(msg); + } catch(Throwable ex) { + //in case errors arise, remove the registration because the subscription + // hasn't get well +// unregister(topic); + throw new MQTTException(ex); + } + + for (String topic : topics) { + unregister(topic); + } - //TODO check the ACK messageID +// register(topic, publishCallback); + updatePinger(); } private void manageSendQoS1(MessageIDMessage msg) throws Throwable{ @@ -371,6 +403,13 @@ public final class Client { m_subscribeBarrier.countDown(); m_receivedSubAckMessageID = messageID; } + + void unsubscribeAckCallback(int messageID) { + LOG.info("unsubscribeAckCallback invoked"); + //NB we share the barrier because in futur will be a single barrier for all + m_subscribeBarrier.countDown(); + m_receivedSubAckMessageID = messageID; + } protected void publishCallback(String topic, byte[] payload) { IPublishCallback callback = m_subscribersList.get(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 7fc689db..74d7723b 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 @@ -7,6 +7,7 @@ import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.proto.messages.SubAckMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -35,6 +36,9 @@ public class ClientMQTTHandler extends IoHandlerAdapter { case SUBACK: handleSubscribeAck(session, (SubAckMessage) msg); break; + case UNSUBACK: + handleUnsubscribeAck(session, (UnsubAckMessage) msg); + break; // case SUBSCRIBE: // handleSubscribe(session, (SubscribeMessage) msg); // break; @@ -57,4 +61,8 @@ public class ClientMQTTHandler extends IoHandlerAdapter { private void handlePublish(IoSession session, PublishMessage pubMessage) { m_callback.publishCallback(pubMessage.getTopicName(), pubMessage.getPayload()); } + + private void handleUnsubscribeAck(IoSession session, UnsubAckMessage unsubAckMessage) { + m_callback.unsubscribeAckCallback(unsubAckMessage.getMessageID()); + } }