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 210a483e..ec3da3ec 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 @@ -21,12 +21,6 @@ public interface IMessaging { void stop(); - /** - * Remove the clientID from topic subscription, if not previously subscribed, - * doesn't reply any error - */ - //void unsubscribe(String topic, String clientID); - void disconnect(IoSession session); void republishStored(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 f9d3919a..a4423586 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 @@ -114,14 +114,6 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_ringBuffer.publish(sequence); } - /* - public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, IoSession session) { - disruptorPublish(new PublishEvent(topic, qos, message, retain, clientID, session)); - } - - public void publish(String topic, byte[] message, QOSType qos, boolean retain, String clientID, int messageID, IoSession session) { - disruptorPublish(new PublishEvent(topic, qos, message, retain, clientID, messageID, session)); - } */ public void subscribe(String clientId, String topic, QOSType qos, boolean cleanSession, int messageID) { Subscription newSubscription = new Subscription(clientId, topic, qos, cleanSession); @@ -129,12 +121,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { disruptorPublish(new SubscribeEvent(newSubscription, messageID)); } - - /*public void unsubscribe(String topic, String clientID) { - disruptorPublish(new UnsubscribeEvent(topic, clientID)); - } */ - - + public void disconnect(IoSession session) { disruptorPublish(new DisconnectEvent(session)); } @@ -422,7 +409,11 @@ public class SimpleMessaging implements IMessaging, EventHandler { // IoSession session = m_clientIDs.get(evt.getSubscription().clientId).getSession(); // session.write(ackMessage).awaitUninterruptibly(); } - + + /** + * Remove the clientID from topic subscription, if not previously subscribed, + * doesn't reply any error + */ protected void processUnsubscribe(IoSession session, String clientID, List topics, int messageID) { LOG.debug("processSubscribe invoked"); 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 802ee138..904b47c1 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 @@ -88,17 +88,6 @@ public class MQTTHandler extends IoHandlerAdapter { private void handleUnsubscribe(IoSession session, UnsubscribeMessage msg) { LOG.info("unregistering the subscriptions"); m_messaging.handleProtocolMessage(session, msg); - - - /*for (String topic : msg.topics()) { - m_messaging.unsubscribe(topic, (String) session.getAttribute(Constants.ATTR_CLIENTID)); - } - //ack the client - UnsubAckMessage ackMessage = new UnsubAckMessage(); - ackMessage.setMessageID(msg.getMessageID()); - - LOG.info("replying with UnsubAck to MSG ID {0}", msg.getMessageID()); - session.write(ackMessage); */ } protected void handlePublish(IoSession session, PublishMessage message) {