diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 22683b5b..ca90217c 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -94,9 +94,9 @@ public class ClientSession { return "ClientSession{clientID='" + clientID + '\'' +"}"; } - public boolean subscribe(String topicFilter, Subscription newSubscription) { - LOG.info("<{}> subscribed to topicFilter <{}> with QoS {}", - newSubscription.getClientId(), topicFilter, + public boolean subscribe(Subscription newSubscription) { + LOG.info("<{}> subscribed to the topic filter <{}> with QoS {}", + newSubscription.getClientId(), newSubscription.getTopicFilter(), AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos())); boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter()); if (!validTopic) { diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index b0fb93da..6981f4ea 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -715,7 +715,7 @@ public class ProtocolProcessor { AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.qos); Subscription newSubscription = new Subscription(clientID, req.topicFilter, qos); - boolean valid = clientSession.subscribe(req.topicFilter, newSubscription); + boolean valid = clientSession.subscribe(newSubscription); ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE); if (valid) { newSubscriptions.add(newSubscription); diff --git a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java new file mode 100644 index 00000000..8ec6074f --- /dev/null +++ b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java @@ -0,0 +1,54 @@ +package io.moquette.spi; + +import io.moquette.parser.proto.messages.AbstractMessage; +import io.moquette.spi.impl.MemoryStorageService; +import io.moquette.spi.impl.subscriptions.Subscription; +import io.moquette.spi.impl.subscriptions.SubscriptionsStore; +import org.junit.Before; +import org.junit.Test; + +import java.util.List; + +import static org.junit.Assert.assertEquals; + +/** + * Created by nedermail on 17/07/16. + */ +public class ClientSessionTest { + + ClientSession session1; + ClientSession session2; + ISessionsStore sessionsStore; + SubscriptionsStore store; + + @Before + public void setUp() { + store = new SubscriptionsStore(); + MemoryStorageService storageService = new MemoryStorageService(); + storageService.initStore(); + this.sessionsStore = storageService.sessionsStore(); + store.init(sessionsStore); + + session1 = sessionsStore.createNewSession("SESSION_ID_1", true); + session2 = sessionsStore.createNewSession("SESSION_ID_2", true); + } + + @Test + public void overridingSubscriptions() { + // Subscribe on /topic with QOSType.MOST_ONE + Subscription oldSubscription = new Subscription(session1.clientID, "/topic", AbstractMessage.QOSType.MOST_ONE); + session1.subscribe(oldSubscription); + store.add(oldSubscription.asClientTopicCouple()); + + // Subscribe on /topic again that overrides the previous subscription. + Subscription overrindingSubscription = new Subscription(session1.clientID, "/topic", AbstractMessage.QOSType.EXACTLY_ONCE); + session1.subscribe(overrindingSubscription); + store.add(overrindingSubscription.asClientTopicCouple()); + + //Verify + List subscriptions = store.matches("/topic"); + assertEquals(1, subscriptions.size()); + Subscription sub = subscriptions.get(0); + assertEquals(overrindingSubscription.getRequestedQos(), sub.getRequestedQos()); + } +} diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java index d21103fb..16457d76 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java @@ -397,24 +397,6 @@ public class SubscriptionsStoreTest { assertEquals(slashSub.getClientId(), remainedSubscription.getClientId()); } - //TODO this test has to be moved on ClientSession, it's not part of subscription logic - @Test - public void overridingSubscriptions() { - ClientSession session = sessionsStore.createNewSession("FAKE_CLI_ID_1", true); - Subscription oldSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE); - session.subscribe("/topic", oldSubscription); - store.add(oldSubscription.asClientTopicCouple()); - Subscription overrindingSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.EXACTLY_ONCE); - session.subscribe("/topic", overrindingSubscription); - store.add(overrindingSubscription.asClientTopicCouple()); - - //Verify - List subscriptions = store.matches("/topic"); - assertEquals(1, subscriptions.size()); - Subscription sub = subscriptions.get(0); - assertEquals(overrindingSubscription.getRequestedQos(), sub.getRequestedQos()); - } - /* Test for Issue #49 * */ @@ -422,15 +404,15 @@ public class SubscriptionsStoreTest { public void duplicatedSubscriptionsWithDifferentQos() { ClientSession session2 = sessionsStore.createNewSession("client2", true); Subscription client2Sub = new Subscription("client2", "client/test/b", AbstractMessage.QOSType.MOST_ONE); - session2.subscribe("client/test/b", client2Sub); + session2.subscribe(client2Sub); store.add(client2Sub.asClientTopicCouple()); ClientSession session1 = sessionsStore.createNewSession("client1", true); Subscription client1SubQoS0 = new Subscription("client1", "client/test/b", AbstractMessage.QOSType.MOST_ONE); - session1.subscribe("client/test/b", client1SubQoS0); + session1.subscribe(client1SubQoS0); store.add(client1SubQoS0.asClientTopicCouple()); Subscription client1SubQoS2 = new Subscription("client1", "client/test/b", AbstractMessage.QOSType.EXACTLY_ONCE); - session1.subscribe("client/test/b", client1SubQoS2); + session1.subscribe(client1SubQoS2); store.add(client1SubQoS2.asClientTopicCouple()); System.out.println(store.dumpTree()); diff --git a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java index a943a65d..77d55062 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -68,12 +68,16 @@ public class MapDBPersistentStoreTest { @Test public void overridingSubscriptions() { - ClientSession session1 = m_sessionsStore.createNewSession("FAKE_CLI_ID_1", true); - Subscription oldSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE); - session1.subscribe("/topic", oldSubscription); - Subscription overrindingSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.EXACTLY_ONCE); - session1.subscribe("/topic", overrindingSubscription); - + ClientSession session1 = m_sessionsStore.createNewSession("SESSION_ID_1", true); + + // Subscribe on /topic with QOSType.MOST_ONE + Subscription oldSubscription = new Subscription(session1.clientID, "/topic", AbstractMessage.QOSType.MOST_ONE); + session1.subscribe(oldSubscription); + + // Subscribe on /topic again that overrides the previous subscription. + Subscription overrindingSubscription = new Subscription(session1.clientID, "/topic", AbstractMessage.QOSType.EXACTLY_ONCE); + session1.subscribe(overrindingSubscription); + //Verify List subscriptions = m_sessionsStore.listAllSubscriptions(); assertEquals(1, subscriptions.size());