From f691767fd50270d40748b7d0af4b5cf92923ba38 Mon Sep 17 00:00:00 2001 From: andrea Date: Sun, 19 Aug 2012 12:17:30 +0200 Subject: [PATCH] Added unsubscribe management for server side --- .../moquette/messaging/spi/IMessaging.java | 6 +++ .../messaging/spi/impl/SimpleMessaging.java | 37 +++++++++----- .../spi/impl/SubscriptionsStore.java | 51 ++++++++++++++++++- .../dna/mqtt/moquette/server/MQTTHandler.java | 20 ++++++++ .../org/dna/mqtt/moquette/server/Server.java | 5 ++ 5 files changed, 105 insertions(+), 14 deletions(-) diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java index 32beebba..0754c059 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java @@ -29,4 +29,10 @@ public interface IMessaging { void removeSubscriptions(String clientID); void close(); + + /** + * Remove the clientID from topic subscription, if not previously subscribed, + * doesn't reply any error + */ + void unsubscribe(String topic, String clientID); } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index 899934e6..d59a5074 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -119,20 +119,33 @@ public class SimpleMessaging implements IMessaging { public void subscribe(String clientId, String topic, QOSType qos) { Subscription newSubscription = new Subscription(clientId, topic, qos); rwLock.writeLock().lock(); - subscriptions.add(newSubscription); - - //scans reatained messages to be published to the new subscription - LOG.debug("Scanning all retained messages..."); - for (Map.Entry entry : m_retainedStore) { - StoredMessage storedMsg = entry.getValue(); - if (matchTopics(entry.getKey(), topic)) { - //fire the as retained the message - m_notifier.notify(newSubscription.clientId, topic, storedMsg.getQos(), - storedMsg.getPayload(), true); + try { + subscriptions.add(newSubscription); + + //scans retained messages to be published to the new subscription + LOG.debug("Scanning all retained messages..."); + for (Map.Entry entry : m_retainedStore) { + StoredMessage storedMsg = entry.getValue(); + if (matchTopics(entry.getKey(), topic)) { + //fire the as retained the message + m_notifier.notify(newSubscription.clientId, topic, storedMsg.getQos(), + storedMsg.getPayload(), true); + } } + } finally { + LOG.debug("Finished firing retained messages"); + rwLock.writeLock().unlock(); + } + } + + + public void unsubscribe(String topic, String clientID) { + rwLock.writeLock().lock(); + try { + subscriptions.removeSubscription(topic, clientID); + } finally { + rwLock.writeLock().unlock(); } - LOG.debug("Finished firing retained messages"); - rwLock.writeLock().unlock(); } 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 bc04ab81..59ffe6bc 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 @@ -217,12 +217,42 @@ public class SubscriptionsStore { } protected void addDirect(Subscription newSubscription) { +// List tokens = new ArrayList(); +// try { +// tokens = splitTopic(newSubscription.topic); +// } catch (ParseException ex) { +// //TODO handle the parse exception +// LOG.error(null, ex); +//// return; +// } +// +// TreeNode current = subscriptions; +// for (Token token : tokens) { +// TreeNode matchingChildren; +// +// //check if a children with the same token already exists +// if ((matchingChildren = current.childWithToken(token)) != null) { +// current = matchingChildren; +// } else { +// //create a new node for the newly inserted token +// matchingChildren = new TreeNode(current); +// matchingChildren.setToken(token); +// current.addChild(matchingChildren); +// current = matchingChildren; +// } +// } + TreeNode current = findMatchingNode(newSubscription.topic); + current.addSubcription(newSubscription); + } + + private TreeNode findMatchingNode(String topic) { List tokens = new ArrayList(); try { - tokens = splitTopic(newSubscription.topic); + tokens = splitTopic(topic); } catch (ParseException ex) { //TODO handle the parse exception LOG.error(null, ex); +// return; } TreeNode current = subscriptions; @@ -240,7 +270,7 @@ public class SubscriptionsStore { current = matchingChildren; } } - current.addSubcription(newSubscription); + return current; } public void add(Subscription newSubscription) { @@ -257,6 +287,23 @@ public class SubscriptionsStore { m_persistent.put(clientID, subs); // pageFile.flush(); } + + public void removeSubscription(String topic, String clientID) { + TreeNode matchNode = findMatchingNode(topic); + + //search fr the subscription to remove + Subscription toBeRemoved = null; + for (Subscription sub : matchNode.subscriptions()) { + if (sub.topic.equals(topic)) { + toBeRemoved = sub; + break; + } + } + + if (toBeRemoved != null) { + matchNode.subscriptions().remove(toBeRemoved); + } + } /** * Visit the topics tree to remove matching subscriptions with clientID diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java index 0c196613..44d1a680 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/MQTTHandler.java @@ -19,6 +19,8 @@ import org.dna.mqtt.moquette.proto.messages.PingRespMessage; import org.dna.mqtt.moquette.proto.messages.PublishMessage; import org.dna.mqtt.moquette.proto.messages.SubAckMessage; import org.dna.mqtt.moquette.proto.messages.SubscribeMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage; +import org.dna.mqtt.moquette.proto.messages.UnsubscribeMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -50,6 +52,9 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { case SUBSCRIBE: handleSubscribe(session, (SubscribeMessage) msg); break; + case UNSUBSCRIBE: + handleUnsubscribe(session, (UnsubscribeMessage) msg); + break; case PUBLISH: handlePublish(session, (PublishMessage) msg); break; @@ -164,6 +169,19 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { LOG.info("replying with SubAct to MSG ID {0}", msg.getMessageID()); session.write(ackMessage); } + + private void handleUnsubscribe(IoSession session, UnsubscribeMessage msg) { + LOG.info("unregistering the subscriptions"); + for (String topic : msg.topics()) { + m_messaging.unsubscribe(topic, (String) session.getAttribute(ATTR_CLIENTID)); + } + //ack the client + UnsubAckMessage ackMessage = new UnsubAckMessage(); + ackMessage.setMessageID(msg.getMessageID()); + + LOG.info("replying with UnsubAct to MSG ID {0}", msg.getMessageID()); + session.write(ackMessage); + } protected void handlePublish(IoSession session, PublishMessage message) { m_messaging.publish(message.getTopicName(), message.getPayload(), @@ -205,6 +223,8 @@ public class MQTTHandler extends IoHandlerAdapter implements INotifier { pubMessage.setTopicName(topic); pubMessage.setQos(qOSType); pubMessage.setPayload(payload); + assert m_clientIDs != null; + assert m_clientIDs.get(clientId) != null; m_clientIDs.get(clientId).getSession().write(pubMessage); } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java index 0326798c..fa3e06ee 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java +++ b/parser/src/main/java/org/dna/mqtt/moquette/server/Server.java @@ -21,10 +21,13 @@ import org.dna.mqtt.moquette.proto.PublishDecoder; import org.dna.mqtt.moquette.proto.PublishEncoder; import org.dna.mqtt.moquette.proto.SubAckEncoder; import org.dna.mqtt.moquette.proto.SubscribeDecoder; +import org.dna.mqtt.moquette.proto.UnsubAckEncoder; +import org.dna.mqtt.moquette.proto.UnsubscribeDecoder; import org.dna.mqtt.moquette.proto.messages.ConnAckMessage; import org.dna.mqtt.moquette.proto.messages.PingRespMessage; 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; /** @@ -52,6 +55,7 @@ public class Server { decoder.addMessageDecoder(new ConnectDecoder()); decoder.addMessageDecoder(new PublishDecoder()); decoder.addMessageDecoder(new SubscribeDecoder()); + decoder.addMessageDecoder(new UnsubscribeDecoder()); decoder.addMessageDecoder(new DisconnectDecoder()); decoder.addMessageDecoder(new PingReqDecoder()); @@ -59,6 +63,7 @@ public class Server { // encoder.addMessageEncoder(ConnectMessage.class, new ConnectEncoder()); encoder.addMessageEncoder(ConnAckMessage.class, new ConnAckEncoder()); encoder.addMessageEncoder(SubAckMessage.class, new SubAckEncoder()); + encoder.addMessageEncoder(UnsubAckMessage.class, new UnsubAckEncoder()); encoder.addMessageEncoder(PublishMessage.class, new PublishEncoder()); encoder.addMessageEncoder(PingRespMessage.class, new PingRespEncoder());