diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java index 6af5505e..ae60785e 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageService.java @@ -225,6 +225,17 @@ public class HawtDBStorageService implements IStorageService { Set 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); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java index 1dcbc2ef..5bc1df59 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/TreeNode.java +++ b/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 { + + 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); } diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageServiceTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/HawtDBStorageServiceTest.java new file mode 100644 index 00000000..5e872195 --- /dev/null +++ b/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 subscriptions = m_storageService.retrieveAllSubscriptions(); + assertEquals(1, subscriptions.size()); + Subscription sub = subscriptions.get(0); + assertEquals(overrindingSubscription.getRequestedQos(), sub.getRequestedQos()); + } +} \ No newline at end of file diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java index 38ed26a2..1e386c73 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/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 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]; diff --git a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java index 0075d0ec..c810fd28 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/server/ServerIntegrationPahoTest.java +++ b/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();