Added remove of subscribed topic during processing of UNSUBSCRIBE message (fix issue #50)

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

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
      */

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<Subscription> 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();

broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java

@@ -587,7 +587,9 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
 
         for (String topic : topics) {
             subscriptions.removeSubscription(topic, clientID);
+            m_sessionsStore.removeSubscription(topic, clientID);
         }
+
         //ack the client
         UnsubAckMessage ackMessage = new UnsubAckMessage();
         ackMessage.setMessageID(messageID);

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<Subscription> 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();

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
      * */