From 90e2277490becf312d805f44ad5838ae2e9b6f91 Mon Sep 17 00:00:00 2001
From: andrea <andrea@samarcanda>
Date: Fri, 30 Aug 2013 15:00:51 +0200
Subject: [PATCH] Moved processUnsubscribe  into ProtocolProcessor

---
 .../messaging/spi/impl/ProtocolProcessor.java | 20 +++++++++++
 .../messaging/spi/impl/SimpleMessaging.java   | 36 +++++++++----------
 2 files changed, 38 insertions(+), 18 deletions(-)

diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
index fafad0af..4aef4ab2 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java
@@ -1,5 +1,6 @@
 package org.dna.mqtt.moquette.messaging.spi.impl;
 
+import java.util.List;
 import java.util.Map;
 import org.apache.mina.core.session.IdleStatus;
 import org.apache.mina.core.session.IoSession;
@@ -18,6 +19,7 @@ import org.dna.mqtt.moquette.proto.messages.PubAckMessage;
 import org.dna.mqtt.moquette.proto.messages.PubRecMessage;
 import org.dna.mqtt.moquette.proto.messages.PubRelMessage;
 import org.dna.mqtt.moquette.proto.messages.PublishMessage;
+import org.dna.mqtt.moquette.proto.messages.UnsubAckMessage;
 import org.dna.mqtt.moquette.server.ConnectionDescriptor;
 import org.dna.mqtt.moquette.server.Constants;
 import org.dna.mqtt.moquette.server.IAuthenticator;
@@ -319,5 +321,23 @@ class ProtocolProcessor {
 //        String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
         subscriptions.deactivate(clientID);
     }
+    
+    /**
+     * Remove the clientID from topic subscription, if not previously subscribed,
+     * doesn't reply any error
+     */
+    void processUnsubscribe(IoSession session, String clientID, List<String> topics, int messageID) {
+        LOG.debug("processSubscribe invoked");
+
+        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);
+    }
 
 }
diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
index 1922019d..dd8a4cbb 100644
--- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
+++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java
@@ -168,7 +168,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
             } else if (message instanceof UnsubscribeMessage) {
                 UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
                 String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
-                processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID());
+                m_processor.processUnsubscribe(session, clientID, unsubMsg.topics(), unsubMsg.getMessageID());
             } else if (message instanceof SubscribeMessage) {
                 String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
                 boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
@@ -245,23 +245,23 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         session.write(ackMessage);
     }
 
-    /**
-     * Remove the clientID from topic subscription, if not previously subscribed,
-     * doesn't reply any error
-     */
-    protected void processUnsubscribe(IoSession session, String clientID, List<String> topics, int messageID) {
-        LOG.debug("processSubscribe invoked");
-
-        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);
-    }
+//    /**
+//     * Remove the clientID from topic subscription, if not previously subscribed,
+//     * doesn't reply any error
+//     */
+//    protected void processUnsubscribe(IoSession session, String clientID, List<String> topics, int messageID) {
+//        LOG.debug("processSubscribe invoked");
+//
+//        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);
+//    }
     
     private void processStop() {
         LOG.debug("processStop invoked");