diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java index 98d60187..e07c33d2 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ClientSession.java @@ -22,6 +22,7 @@ import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.ArrayList; import java.util.HashSet; import java.util.List; import java.util.Set; @@ -84,4 +85,15 @@ public class ClientSession { m_sessionsStore.addNewSubscription(newSubscription); return true; } + + public void unsubscribeFrom(String topicFilter) { + m_sessionsStore.removeSubscription(topicFilter, clientID); + Set subscriptionsToRemove = new HashSet<>(); + for (Subscription sub : this.subscriptions) { + if (sub.getTopicFilter().equals(topicFilter)) { + subscriptionsToRemove.add(sub); + } + } + subscriptions.removeAll(subscriptionsToRemove); + } } \ No newline at end of file diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index fca6e847..c3b41c80 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -568,6 +568,8 @@ public class ProtocolProcessor { String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); LOG.debug("UNSUBSCRIBE subscription on topics {} for clientID <{}>", topics, clientID); + ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); + for (String topic : topics) { boolean validTopic = SubscriptionsStore.validate(topic); if (!validTopic) { @@ -578,6 +580,7 @@ public class ProtocolProcessor { } subscriptions.removeSubscription(topic, clientID); + clientSession.unsubscribeFrom(topic); m_interceptor.notifyTopicUnsubscribed(topic, clientID); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java index 3e80b622..17b3fff6 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -198,7 +198,6 @@ public class SubscriptionsStore { } //spin lock repeating till we can, swap root, if can't swap just re-do the operation } while(!subscriptions.compareAndSet(oldRoot, couple.root)); - m_sessionsStore.removeSubscription(topic, clientID); } /**