Enabled overriding of subscription from same client to same topic with different QoS, the latest wins

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java

@@ -225,6 +225,17 @@ public class HawtDBStorageService implements IStorageService {
         Set<Subscription> subs = m_persistentSubscriptions.get(clientID);
         if (!subs.contains(newSubscription)) {
             LOG.debug("updating clientID {} subscriptions set with new subscription", clientID);
+            //TODO check the subs doesn't contain another subscription to the same topic with different
+            Subscription existingSubscription = null;
+            for (Subscription scanSub : subs) {
+                if (newSubscription.getTopic().equals(scanSub.getTopic())) {
+                    existingSubscription = scanSub;
+                    break;
+                }
+            }
+            if (existingSubscription != null) {
+                subs.remove(existingSubscription);
+            }
             subs.add(newSubscription);
             m_persistentSubscriptions.put(clientID, subs);
             LOG.debug("clientID {} subscriptions set now is {}", clientID, subs);

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java

@@ -1,11 +1,21 @@
 package org.dna.mqtt.moquette.messaging.spi.impl.subscriptions;
 
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.LinkedBlockingQueue;
 
 class TreeNode {
+    
+    private class ClientIDComparator implements Comparator<Subscription> {
+
+        public int compare(Subscription o1, Subscription o2) {
+            return o1.getClientId().compareTo(o2.getClientId());
+        }
+        
+    }
 
     TreeNode m_parent;
     Token m_token;
@@ -25,10 +35,16 @@ class TreeNode {
     }
 
     void addSubscription(Subscription s) {
-        //avoid double registering
+        //avoid double registering for same clientID, topic and QoS
         if (m_subscriptions.contains(s)) {
             return;
         }
+        //remove existing subscription for same client and topic but different QoS
+        int existingSubIdx = Collections.binarySearch(m_subscriptions, s, new ClientIDComparator());
+        if (existingSubIdx >= 0) {
+            m_subscriptions.remove(existingSubIdx);
+        }
+        
         m_subscriptions.add(s);
     }
 

broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageServiceTest.java

@@ -0,0 +1,52 @@
+package org.dna.mqtt.moquette.messaging.spi.impl;
+
+import java.io.File;
+import java.util.List;
+import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription;
+import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
+import org.dna.mqtt.moquette.server.Server;
+import org.junit.After;
+import org.junit.Test;
+import static org.junit.Assert.*;
+import org.junit.Before;
+
+/**
+ *
+ * @author andrea
+ */
+public class HawtDBStorageServiceTest {
+
+    HawtDBStorageService m_storageService;
+        
+    @Before
+    public void setUp() throws Exception {
+        File dbFile = new File(Server.STORAGE_FILE_PATH);
+        assertFalse(String.format("The DB storagefile %s already exists", Server.STORAGE_FILE_PATH), dbFile.exists());
+        
+        m_storageService = new HawtDBStorageService();
+        m_storageService.initStore();
+    }
+
+    @After
+    public void tearDown() {
+        File dbFile = new File(Server.STORAGE_FILE_PATH);
+        if (dbFile.exists()) {
+            dbFile.delete();
+        }
+        assertFalse(dbFile.exists());
+    }
+
+    @Test
+    public void overridingSubscriptions() {
+        Subscription oldSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false);
+        m_storageService.addNewSubscription(oldSubscription, oldSubscription.getClientId());
+        Subscription overrindingSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.EXACTLY_ONCE, false);
+        m_storageService.addNewSubscription(overrindingSubscription, overrindingSubscription.getClientId());
+        
+        //Verify
+        List<Subscription> subscriptions = m_storageService.retrieveAllSubscriptions();
+        assertEquals(1, subscriptions.size());
+        Subscription sub = subscriptions.get(0);
+        assertEquals(overrindingSubscription.getRequestedQos(), sub.getRequestedQos());
+    }
+}

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

@@ -345,6 +345,20 @@ public class SubscriptionsStoreTest {
         Subscription remainedSubscription = aStore.matches("/topic").get(0);
         assertEquals(slashSub.getClientId(), remainedSubscription.getClientId());
     }
+    
+    @Test
+    public void overridingSubscriptions() {
+        Subscription oldSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false);
+        store.add(oldSubscription);
+        Subscription overrindingSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.EXACTLY_ONCE, false);
+        store.add(overrindingSubscription);
+        
+        //Verify
+        List<Subscription> subscriptions = store.matches("/topic");
+        assertEquals(1, subscriptions.size());
+        Subscription sub = subscriptions.get(0);
+        assertEquals(overrindingSubscription.getRequestedQos(), sub.getRequestedQos());
+    }
 
     private static Token[] asArray(Object... l) {
         Token[] tokens = new Token[l.length];

broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java

@@ -87,7 +87,7 @@ public class ServerIntegrationPahoTest {
     @Before
     public void setUp() throws Exception {
         File dbFile = new File(Server.STORAGE_FILE_PATH);
-        assertFalse(dbFile.exists());
+        assertFalse(String.format("The DB storagefile %s already exists", Server.STORAGE_FILE_PATH), dbFile.exists());
         
         startServer();