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 34e769ff..210a483e 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 @@ -25,7 +25,7 @@ public interface IMessaging { * Remove the clientID from topic subscription, if not previously subscribed, * doesn't reply any error */ - void unsubscribe(String topic, String clientID); + //void unsubscribe(String topic, String clientID); void disconnect(IoSession session); 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 9bce5851..f9d3919a 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 @@ -14,6 +14,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import org.apache.mina.core.session.IdleStatus; import org.apache.mina.core.session.IoSession; +import org.apache.mina.filter.util.SessionAttributeInitializingFilter; import org.dna.mqtt.moquette.messaging.spi.IMatchingCondition; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.IStorageService; @@ -129,9 +130,9 @@ public class SimpleMessaging implements IMessaging, EventHandler { } - public void unsubscribe(String topic, String clientID) { + /*public void unsubscribe(String topic, String clientID) { disruptorPublish(new UnsubscribeEvent(topic, clientID)); - } + } */ public void disconnect(IoSession session) { @@ -205,9 +206,9 @@ public class SimpleMessaging implements IMessaging, EventHandler { processPublish((PublishEvent) evt); } else if (evt instanceof SubscribeEvent) { processSubscribe((SubscribeEvent) evt); - } else if (evt instanceof UnsubscribeEvent) { + } /*else if (evt instanceof UnsubscribeEvent) { processUnsubscribe((UnsubscribeEvent) evt); - } else if (evt instanceof StopEvent) { + } */else if (evt instanceof StopEvent) { processStop(); } else if (evt instanceof DisconnectEvent) { DisconnectEvent disEvt = (DisconnectEvent) evt; @@ -247,6 +248,11 @@ public class SimpleMessaging implements IMessaging, EventHandler { //close the TCP connection //session.close(true); processDisconnect(session, clientID); + } else if (message instanceof UnsubscribeMessage) { + UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message; + String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID); + processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID()); + } else { throw new RuntimeException("Illegal message received " + message); } @@ -417,9 +423,18 @@ public class SimpleMessaging implements IMessaging, EventHandler { // session.write(ackMessage).awaitUninterruptibly(); } - protected void processUnsubscribe(UnsubscribeEvent evt) { + protected void processUnsubscribe(IoSession session, String clientID, List topics, int messageID) { LOG.debug("processSubscribe invoked"); - subscriptions.removeSubscription(evt.getTopic(), evt.getClientID()); + + for (String topic : topics) { + subscriptions.removeSubscription(topic, clientID); + } + //ack the client + UnsubAckMessage ackMessage = new UnsubAckMessage(); + ackMessage.setMessageID(messageID); + + LOG.info("replying with UnsubAck to MSG ID {0}", messageID); + session.write(ackMessage); } protected void processRemoveAllSubscriptions(String 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 0e7aa269..802ee138 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 @@ -87,7 +87,10 @@ public class MQTTHandler extends IoHandlerAdapter { private void handleUnsubscribe(IoSession session, UnsubscribeMessage msg) { LOG.info("unregistering the subscriptions"); - for (String topic : msg.topics()) { + m_messaging.handleProtocolMessage(session, msg); + + + /*for (String topic : msg.topics()) { m_messaging.unsubscribe(topic, (String) session.getAttribute(Constants.ATTR_CLIENTID)); } //ack the client @@ -95,7 +98,7 @@ public class MQTTHandler extends IoHandlerAdapter { ackMessage.setMessageID(msg.getMessageID()); LOG.info("replying with UnsubAck to MSG ID {0}", msg.getMessageID()); - session.write(ackMessage); + session.write(ackMessage); */ } protected void handlePublish(IoSession session, PublishMessage message) {