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 1d19009a..c094bc14 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,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); } 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 170e1045..59be5b38 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 @@ -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(); + } } diff --git a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java b/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java index d9a5f219..682c9d9e 100644 --- a/parser/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStore.java +++ b/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 subsToRemove = new ArrayList(); + 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 matches(String topic) { List tokens = new ArrayList(); diff --git a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java b/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java index d45670c6..039156ce 100644 --- a/parser/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/SubscriptionsStoreTest.java +++ b/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) {