From 336c452ef3c63292012eca0a07757e12a55cbbdc Mon Sep 17 00:00:00 2001 From: andrea Date: Sun, 21 Jun 2015 17:48:28 +0200 Subject: [PATCH] Added remove of subscribed topic during processing of UNSUBSCRIBE message (fix issue #50) --- ChangeLog.txt | 3 +- .../eclipse/moquette/spi/ISessionsStore.java | 5 +++ .../spi/impl/MemoryStorageService.java | 21 ++++++++++ .../moquette/spi/impl/ProtocolProcessor.java | 2 + .../spi/persistence/MapDBPersistentStore.java | 23 +++++++++++ .../server/ServerRestartIntegrationTest.java | 38 +++++++++++++++++++ 6 files changed, 91 insertions(+), 1 deletion(-) diff --git a/ChangeLog.txt b/ChangeLog.txt index 5385be2d..8819482b 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -16,7 +16,8 @@ Version 0.7: Removed cause of leakage of Netty Sessions (fix issue #39) Cleaned Netty's stuff reference from Events used in Disruptor (fix issue #40) Bad assignment of PacketID during publish to subscribed QoS0 topics (fix issue #38) - Fixed duplicated subscription with different QoS same client and topic (fix issue # + Fixed duplicated subscription with different QoS same client and topic (fix issue #49) + Fixed UNSUBSCRIBE that missed to remove the subscription from the persistent store (fix issue #50) Version 0.6: Fix issue #42, bad password file path composition diff --git a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java index 770cd524..77c3153e 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/ISessionsStore.java @@ -32,6 +32,11 @@ public interface ISessionsStore { */ void addNewSubscription(Subscription newSubscription); + /** + * Removed a specific subscription + * */ + void removeSubscription(String topic, String clientID); + /** * Remove all the subscriptions of the session */ diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java index 7e836d5d..3c90e13f 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/MemoryStorageService.java @@ -158,6 +158,27 @@ public class MemoryStorageService implements IMessagesStore, ISessionsStore { return nextPacketId; } + @Override + public void removeSubscription(String topic, String clientID) { + LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); + if (!m_persistentSubscriptions.containsKey(clientID)) { + return; + } + Set clientSubscriptions = m_persistentSubscriptions.get(clientID); + //search for the subscription to remove + Subscription toBeRemoved = null; + for (Subscription sub : clientSubscriptions) { + if (sub.getTopicFilter().equals(topic)) { + toBeRemoved = sub; + break; + } + } + + if (toBeRemoved != null) { + clientSubscriptions.remove(toBeRemoved); + } + } + @Override public void addNewSubscription(Subscription newSubscription) { final String clientID = newSubscription.getClientId(); 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 2364051c..dd85f07e 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 @@ -587,7 +587,9 @@ class ProtocolProcessor implements EventHandler { for (String topic : topics) { subscriptions.removeSubscription(topic, clientID); + m_sessionsStore.removeSubscription(topic, clientID); } + //ack the client UnsubAckMessage ackMessage = new UnsubAckMessage(); ackMessage.setMessageID(messageID); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java index 09627a65..5771b87b 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/persistence/MapDBPersistentStore.java @@ -221,6 +221,29 @@ public class MapDBPersistentStore implements IMessagesStore, ISessionsStore { return nextPacketId; } + @Override + public void removeSubscription(String topic, String clientID) { + LOG.debug("removeSubscription topic filter: {} for clientID: {}", topic, clientID); + if (!m_persistentSubscriptions.containsKey(clientID)) { + return; + } + Set clientSubscriptions = m_persistentSubscriptions.get(clientID); + //search for the subscription to remove + Subscription toBeRemoved = null; + for (Subscription sub : clientSubscriptions) { + if (sub.getTopicFilter().equals(topic)) { + toBeRemoved = sub; + break; + } + } + + if (toBeRemoved != null) { + clientSubscriptions.remove(toBeRemoved); + } + m_persistentSubscriptions.put(clientID, clientSubscriptions); + m_db.commit(); + } + public void addNewSubscription(Subscription newSubscription) { LOG.debug("addNewSubscription invoked with subscription {}", newSubscription); final String clientID = newSubscription.getClientId(); diff --git a/broker/src/test/java/org/eclipse/moquette/server/ServerRestartIntegrationTest.java b/broker/src/test/java/org/eclipse/moquette/server/ServerRestartIntegrationTest.java index 52be3a8b..6c2a576a 100644 --- a/broker/src/test/java/org/eclipse/moquette/server/ServerRestartIntegrationTest.java +++ b/broker/src/test/java/org/eclipse/moquette/server/ServerRestartIntegrationTest.java @@ -140,6 +140,44 @@ public class ServerRestartIntegrationTest { m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), QoS.AT_MOST_ONCE, false); } + @Test + public void testClientDoesntRemainSubscribedAfterASubscriptionAndServerRestart() throws Exception { + m_mqtt.setClientId("Subscriber"); + //subscribe to /topic + m_subscriber = m_mqtt.blockingConnection(); + m_subscriber.connect(); + Topic[] topics = new Topic[]{new Topic("/topic", QoS.AT_MOST_ONCE)}; + //subscribe /topic + m_subscriber.subscribe(topics); + //unsubscribe from /topic + m_subscriber.unsubscribe(new String[]{"/topic"}); + m_subscriber.disconnect(); + + //shutdown the server + m_server.stopServer(); + + //restart the server + m_server.startServer(IntegrationUtils.prepareTestPropeties()); + //subscriber reconnects + MQTT mqttSub = new MQTT(); + mqttSub.setHost("localhost", 1883); + mqttSub.setClientId("Subscriber"); + m_subscriber = mqttSub.blockingConnection(); + m_subscriber.connect(); + + //publisher publishes on /topic + MQTT mqtt = new MQTT(); + mqtt.setHost("localhost", 1883); + mqtt.setClientId("Publisher"); + m_publisher = mqtt.blockingConnection(); + m_publisher.connect(); + m_publisher.publish("/topic", "Hello world MQTT!!".getBytes(), QoS.AT_LEAST_ONCE, false); + + //Expected + //the subscriber doesn't get notified (it's fully unsubscribed) + assertNull(m_subscriber.receive(1, TimeUnit.SECONDS)); + } + /** * Connect subscribe to topic and publish on the same topic * */