Modified the subscriptions storage to removed the subscriptions for a certain clientID

parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java

@@ -21,4 +21,10 @@ public interface IMessaging {
      * Subscribe a client to a specified topic with a defined level
      */
     void subscribe(String clientId, String topic, QOSType qos);
+
+    /**
+     * Remove all subscription to any topic the client (identified by clientID)
+     * was subscribed
+     */
+    void removeSubscriptions(String clientID);
 }

parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java

@@ -47,4 +47,10 @@ public class SimpleMessaging implements IMessaging {
     protected SubscriptionsStore getSubscriptions() {
         return subscriptions;
     }
+
+    public void removeSubscriptions(String clientID) {
+        rwLock.writeLock().lock();
+        subscriptions.removeForClient(clientID);
+        rwLock.writeLock().unlock();
+    }
 }

parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java

@@ -15,7 +15,7 @@ import java.util.logging.Logger;
  * @author andrea
  */
 public class SubscriptionsStore {
-    
+
     private class TreeNode {
         TreeNode m_parent;
         Token m_token;
@@ -111,6 +111,25 @@ public class SubscriptionsStore {
             }
             return res;
         }
+
+        void removeClientSubscriptions(String clientID) {
+            //collect what to delete and then delete to avoid ConcurrentModification
+            List<Subscription> subsToRemove = new ArrayList<Subscription>();
+            for(Subscription s : m_subscriptions) {
+                if (s.clientId.equals(clientID)) {
+                    subsToRemove.add(s);
+                }
+            }
+            
+            for (Subscription s : subsToRemove) {
+               m_subscriptions.remove(s); 
+            }
+            
+            //go deep
+            for (TreeNode child : m_children) {
+                child.removeClientSubscriptions(clientID);
+            }
+        }
     }
     
     protected static class Token {
@@ -198,6 +217,13 @@ public class SubscriptionsStore {
         }
         current.addSubcription(newSubscription);
     }
+    
+    /**
+     * Visit the topics tree to remove matching subscriptions with clientID
+     */
+    public void removeForClient(String clientID) {
+        subscriptions.removeClientSubscriptions(clientID);
+    }
 
     public List<Subscription> matches(String topic) {
         List<Token> tokens = new ArrayList<Token>();

parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java

@@ -185,6 +185,31 @@ public class SubscriptionsStoreTest {
         store.add(new Subscription(null, "finance/#", AbstractMessage.QOSType.MOST_ONE));
         assertFalse(store.matches("finance").isEmpty());
     }
+    
+    
+    @Test
+    public void testRemoveClientSubscriptions_existingClientID() {
+        String cliendID = "FAKE_CLID_1";
+        store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE));
+        
+        //Exercise
+        store.removeForClient(cliendID);
+        
+        //Verify
+        assertEquals(0, store.size());
+    }
+    
+    @Test
+    public void testRemoveClientSubscriptions_notexistingClientID() {
+        String cliendID = "FAKE_CLID_1";
+        store.add(new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE));
+        
+        //Exercise
+        store.removeForClient("FAKE_CLID_2");
+        
+        //Verify
+        assertEquals(1, store.size());
+    }
 
 
     private static Token[] asArray(Object... l) {