From ff0139d809be10f0d7a30827276ee26c4249933f Mon Sep 17 00:00:00 2001
From: andrea <andrea@samarcanda>
Date: Fri, 30 Aug 2013 18:55:11 +0200
Subject: [PATCH] Changed some methods visibility

---
 .../messaging/spi/impl/ProtocolProcessor.java | 10 +++++---
 .../messaging/spi/impl/SimpleMessaging.java   | 23 ++++---------------
 2 files changed, 12 insertions(+), 21 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 8c82a789..8e075ab9 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
@@ -140,7 +140,7 @@ class ProtocolProcessor {
         session.write(okResp);
     }
     
-    void processRemoveAllSubscriptions(String clientID) {
+    private void processRemoveAllSubscriptions(String clientID) {
         LOG.debug("processRemoveAllSubscriptions invoked");
         subscriptions.removeForClient(clientID);
 
@@ -185,7 +185,7 @@ class ProtocolProcessor {
     /**
      * Flood the subscribers with the message to notify. MessageID is optional and should only used for QoS 1 and 2
      * */
-    void publish2Subscribers(String topic, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) {
+    private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) {
         for (final Subscription sub : subscriptions.matches(topic)) {
             if (qos == AbstractMessage.QOSType.MOST_ONE) {
                 //QoS 0
@@ -316,7 +316,11 @@ class ProtocolProcessor {
         m_storageService.cleanInFlight(publishKey);
     }
     
-    void processDisconnect(IoSession session, String clientID) throws InterruptedException {
+    void processDisconnect(IoSession session, String clientID, boolean cleanSession) throws InterruptedException {
+        if (cleanSession) {
+            //cleanup topic subscriptions
+            processRemoveAllSubscriptions(clientID);
+        }
 //        m_notifier.disconnect(evt.getSession());
         m_clientIDs.remove(clientID);
         session.close(true);
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 9dca827f..07fe461f 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
@@ -88,9 +88,9 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     }
 
     //method used by hte Notifier to re-put an event on the inbound queue
-    private void refill(MessagingEvent evt) {
-        disruptorPublish(evt);
-    }
+//    private void refill(MessagingEvent evt) {
+//        disruptorPublish(evt);
+//    }
 
     public void republishStored(String clientID) {
         //create the event to push
@@ -102,15 +102,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         disruptorPublish(new ProtocolEvent(session, msg));
     }
 
-
-    /**
-     * NOT SAFE Method, to be removed because used only in tests
-     */
-    protected SubscriptionsStore getSubscriptions() {
-        return subscriptions;
-    }
-
-
     public void stop() {
         disruptorPublish(new StopEvent());
     }
@@ -125,7 +116,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         } else if (evt instanceof DisconnectEvent) {
             DisconnectEvent disEvt = (DisconnectEvent) evt;
             String clientID = (String) disEvt.getSession().getAttribute(Constants.ATTR_CLIENTID);
-            m_processor.processDisconnect(disEvt.getSession(), clientID);
+            m_processor.processDisconnect(disEvt.getSession(), clientID, false);
         } else if (evt instanceof RepublishEvent) {
             processRepublish((RepublishEvent) evt);
         } else if (evt instanceof ProtocolEvent) {
@@ -149,14 +140,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
             } else if (message instanceof DisconnectMessage) {
                 String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);
                 boolean cleanSession = (Boolean) session.getAttribute(Constants.CLEAN_SESSION);
-                if (cleanSession) {
-                    //cleanup topic subscriptions
-                    m_processor.processRemoveAllSubscriptions(clientID);
-                }
 
                 //close the TCP connection
                 //session.close(true);
-                m_processor.processDisconnect(session, clientID);
+                m_processor.processDisconnect(session, clientID, cleanSession);
             } else if (message instanceof UnsubscribeMessage) {
                 UnsubscribeMessage unsubMsg = (UnsubscribeMessage) message;
                 String clientID = (String) session.getAttribute(Constants.ATTR_CLIENTID);