From 5a1cd97b717daad6a25d84aece50b5e82b9671de Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 6 Dec 2015 19:55:04 +0100 Subject: [PATCH] Removed clean session flag from subscription and moved in the correct place (ClientSession) --- .../java/io/moquette/spi/ClientSession.java | 4 ++ .../moquette/spi/impl/ProtocolProcessor.java | 7 +-- .../spi/impl/subscriptions/Subscription.java | 13 +---- .../spi/impl/BrokerInterceptorTest.java | 2 +- .../spi/impl/ProtocolProcessorTest.java | 29 +++++----- .../impl/ProtocolProcessor_CONNECT_Test.java | 3 +- .../subscriptions/SubscriptionsStoreTest.java | 58 +++++++++---------- .../persistence/MapDBPersistentStoreTest.java | 4 +- 8 files changed, 56 insertions(+), 64 deletions(-) diff --git a/broker/src/main/java/io/moquette/spi/ClientSession.java b/broker/src/main/java/io/moquette/spi/ClientSession.java index 2357b55a..e977ef20 100644 --- a/broker/src/main/java/io/moquette/spi/ClientSession.java +++ b/broker/src/main/java/io/moquette/spi/ClientSession.java @@ -142,6 +142,10 @@ public class ClientSession { messagesStore.dropMessagesInSession(this.clientID); } + public boolean isCleanSession() { + return this.cleanSession; + } + public void activate() { this.active = true; this.m_sessionsStore.activate(this.clientID); 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 919a7e3e..390a420d 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -26,7 +26,6 @@ import io.moquette.spi.ClientSession; import io.moquette.spi.IMatchingCondition; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; -import io.moquette.spi.ISessionsStore.ClientTopicCouple; import io.moquette.spi.impl.security.IAuthenticator; import io.moquette.spi.impl.security.IAuthorizator; import io.moquette.spi.impl.subscriptions.SubscriptionsStore; @@ -384,8 +383,7 @@ public class ProtocolProcessor { } else { //QoS 1 or 2 //if the target subscription is not clean session and is not connected => store it - //TODO isCleanSession MUST be on the targetSession not the Subscription - if (!sub.isCleanSession() && !targetSession.isActive()) { + if (!targetSession.isCleanSession() && !targetSession.isActive()) { //store the message in targetSession queue to deliver targetSession.enqueueToDeliver(guid); } else { @@ -594,7 +592,6 @@ public class ProtocolProcessor { public void processSubscribe(ServerChannel session, SubscribeMessage msg) { String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); - boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION); LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID()); ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); @@ -605,7 +602,7 @@ public class ProtocolProcessor { List newSubscriptions = new ArrayList<>(); for (SubscribeMessage.Couple req : msg.subscriptions()) { AbstractMessage.QOSType qos = AbstractMessage.QOSType.valueOf(req.getQos()); - Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos, cleanSession); + Subscription newSubscription = new Subscription(clientID, req.getTopicFilter(), qos); //boolean valid = subscribeSingleTopic(newSubscription, req.getTopicFilter()); boolean valid = clientSession.subscribe(req.getTopicFilter(), newSubscription); ackMessage.addType(valid ? qos : AbstractMessage.QOSType.FAILURE); diff --git a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java index 40fdd716..e9965d3d 100644 --- a/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java +++ b/broker/src/main/java/io/moquette/spi/impl/subscriptions/Subscription.java @@ -31,21 +31,18 @@ public class Subscription implements Serializable { QOSType requestedQos; //max QoS acceptable String clientId; String topicFilter; - boolean cleanSession; boolean active = true; - public Subscription(String clientId, String topicFilter, QOSType requestedQos, boolean cleanSession) { + public Subscription(String clientId, String topicFilter, QOSType requestedQos) { this.requestedQos = requestedQos; this.clientId = clientId; this.topicFilter = topicFilter; - this.cleanSession = cleanSession; } public Subscription(Subscription orig) { this.requestedQos = orig.requestedQos; this.clientId = orig.clientId; this.topicFilter = orig.topicFilter; - this.cleanSession = orig.cleanSession; this.active = orig.active; } @@ -61,14 +58,6 @@ public class Subscription implements Serializable { return topicFilter; } - /** - * @deprecated this concept must be moved inside ClientSession and not subscription. - * */ - @Deprecated - public boolean isCleanSession() { - return this.cleanSession; - } - @Override public boolean equals(Object o) { if (this == o) return true; diff --git a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java index b64bf7cd..9f2901de 100644 --- a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java @@ -106,7 +106,7 @@ public class BrokerInterceptorTest { @Test public void testNotifyTopicSubscribed() throws Exception { - interceptor.notifyTopicSubscribed(new Subscription("cli1", "o2", AbstractMessage.QOSType.MOST_ONE, true)); + interceptor.notifyTopicSubscribed(new Subscription("cli1", "o2", AbstractMessage.QOSType.MOST_ONE)); interval(); assertEquals(70, n.get()); } diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 8fa91253..8de4865e 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -97,7 +97,7 @@ public class ProtocolProcessorTest { @Test public void testPublish() throws InterruptedException { final Subscription subscription = new Subscription(FAKE_CLIENT_ID, - FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true); + FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); //subscriptions.matches(topic) redefine the method to return true SubscriptionsStore subs = new SubscriptionsStore() { @@ -120,7 +120,7 @@ public class ProtocolProcessorTest { connectMessage.setProtocolVersion((byte) 3); connectMessage.setClientID(FAKE_CLIENT_ID); m_sessionStore.createNewSession(FAKE_CLIENT_ID, false); - connectMessage.setCleanSession(subscription.isCleanSession()); + connectMessage.setCleanSession(true); m_processor.processConnect(m_session, connectMessage); @@ -142,9 +142,9 @@ public class ProtocolProcessorTest { @Test public void testPublishToMultipleSubscribers() throws InterruptedException { final Subscription subscription = new Subscription(FAKE_CLIENT_ID, - FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true); + FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); final Subscription subscriptionClient2 = new Subscription(FAKE_CLIENT_ID2, - FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true); + FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); //subscriptions.matches(topic) redefine the method to return true SubscriptionsStore subs = new SubscriptionsStore() { @@ -168,7 +168,7 @@ public class ProtocolProcessorTest { ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProtocolVersion((byte) 3); connectMessage.setClientID(FAKE_CLIENT_ID); - connectMessage.setCleanSession(subscription.isCleanSession()); + connectMessage.setCleanSession(true); m_processor.processConnect(firstReceiverSession, connectMessage); //connect the second fake subscriber @@ -176,7 +176,8 @@ public class ProtocolProcessorTest { ConnectMessage connectMessage2 = new ConnectMessage(); connectMessage2.setProtocolVersion((byte) 3); connectMessage2.setClientID(FAKE_CLIENT_ID2); - connectMessage2.setCleanSession(subscription.isCleanSession()); + connectMessage2.setCleanSession(true); + connectMessage2.setCleanSession(true); m_processor.processConnect(secondReceiverSession, connectMessage2); //Exercise @@ -215,7 +216,7 @@ public class ProtocolProcessorTest { //Verify assertTrue(m_session.getReceivedMessage() instanceof SubAckMessage); - Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false); + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); assertTrue(subscriptions.contains(expectedSubscription)); } @@ -235,7 +236,7 @@ public class ProtocolProcessorTest { //Verify assertEquals(1, subscriptions.size()); - Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false); + Subscription expectedSubscription = new Subscription(FAKE_CLIENT_ID, FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); assertTrue(subscriptions.contains(expectedSubscription)); } @@ -304,7 +305,7 @@ public class ProtocolProcessorTest { //simulate a connect that register a clientID to an IoSession final Subscription subscription = new Subscription(FAKE_PUBLISHER_ID, - FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, true); + FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE); //subscriptions.matches(topic) redefine the method to return true SubscriptionsStore subs = new SubscriptionsStore() { @@ -326,7 +327,7 @@ public class ProtocolProcessorTest { ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setClientID(FAKE_PUBLISHER_ID); connectMessage.setProtocolVersion((byte) 3); - connectMessage.setCleanSession(subscription.isCleanSession()); + connectMessage.setCleanSession(true); m_processor.processConnect(m_session, connectMessage); ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); PublishMessage pubmsg = new PublishMessage(); @@ -382,7 +383,7 @@ public class ProtocolProcessorTest { m_sessionStore.createNewSession("Subscriber", false).deactivate(); SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); - Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE, false); + Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE); List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); @@ -408,7 +409,7 @@ public class ProtocolProcessorTest { //create an inactive session for Subscriber m_sessionStore.createNewSession("Subscriber", false).deactivate(); SubscriptionsStore mockedSubscriptions = mock(SubscriptionsStore.class); - Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE, true); + Subscription inactiveSub = new Subscription("Subscriber", "/topic", QOSType.LEAST_ONE); List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); @@ -482,8 +483,8 @@ public class ProtocolProcessorTest { sessionsStore.createNewSession("Sub A", false).activate(); sessionsStore.createNewSession("Sub B", false).activate(); - Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE, false); - Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE, false); + Subscription subQos1 = new Subscription("Sub A", "a/b", QOSType.LEAST_ONE); + Subscription subQos2 = new Subscription("Sub B", "a/+", QOSType.EXACTLY_ONCE); sessionsStore.addNewSubscription(subQos1); sessionsStore.addNewSubscription(subQos2); SubscriptionsStore subscriptions = new SubscriptionsStore(); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 399937ef..85a2f8e3 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java @@ -276,7 +276,8 @@ public class ProtocolProcessor_CONNECT_Test { m_session.setAttribute(NettyChannel.ATTR_KEY_CLIENTID, "CliID"); m_session.setAttribute(NettyChannel.ATTR_KEY_CLEANSESSION, false); m_processor.processSubscribe(m_session, subscribeMsg); - Subscription expectedSubscription = new Subscription("CliID", ProtocolProcessorTest.FAKE_TOPIC, AbstractMessage.QOSType.MOST_ONE, false); + Subscription expectedSubscription = new Subscription("CliID", ProtocolProcessorTest.FAKE_TOPIC, + AbstractMessage.QOSType.MOST_ONE); assertTrue(subscriptions.contains(expectedSubscription)); //disconnect 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 e6ace70a..f6d52fe4 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 @@ -106,12 +106,12 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimple() { - Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE, false); + Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(slashSub); store.add(slashSub.asClientTopicCouple()); assertTrue(store.matches("finance").isEmpty()); - Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE, false); + Subscription slashFinanceSub = new Subscription("FAKE_CLI_ID_1", "/finance", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(slashFinanceSub); store.add(slashFinanceSub.asClientTopicCouple()); assertTrue(store.matches("finance").isEmpty()); @@ -122,12 +122,12 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti() { - Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(anySub); store.add(anySub.asClientTopicCouple()); assertTrue(store.matches("finance").contains(anySub)); - Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); + Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", "finance/#", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(financeAnySub); store.add(financeAnySub.asClientTopicCouple()); assertTrue(store.matches("finance").containsAll(Arrays.asList(financeAnySub, anySub))); @@ -135,8 +135,8 @@ public class SubscriptionsStoreTest { @Test public void testMatchingDeepMulti_one_layer() { - Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false); - Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE); + Subscription financeAnySub = new Subscription("FAKE_CLI_ID_2", "finance/#", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(anySub); sessionsStore.addNewSubscription(financeAnySub); store.add(anySub.asClientTopicCouple()); @@ -150,7 +150,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchingDeepMulti_two_layer() { - Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE, false); + Subscription financeAnySub = new Subscription("FAKE_CLI_ID_1", "finance/stock/#", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(financeAnySub); store.add(financeAnySub.asClientTopicCouple()); @@ -160,12 +160,12 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleSingle() { - Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE, false); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(anySub); store.add(anySub.asClientTopicCouple()); assertTrue(store.matches("finance").contains(anySub)); - Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE, false); + Subscription financeOne = new Subscription("FAKE_CLI_ID_1", "finance/+", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(financeOne); store.add(financeOne.asClientTopicCouple()); assertTrue(store.matches("finance/stock").contains(financeOne)); @@ -173,7 +173,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchManySingle() { - Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE, false); + Subscription manySub = new Subscription("FAKE_CLI_ID_1", "+/+", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(manySub); store.add(manySub.asClientTopicCouple()); @@ -184,10 +184,10 @@ public class SubscriptionsStoreTest { @Test public void testMatchSlashSingle() { - Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE, false); + Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/+", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(slashPlusSub); store.add(slashPlusSub.asClientTopicCouple()); - Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE, false); + Subscription anySub = new Subscription("FAKE_CLI_ID_1", "+", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(anySub); store.add(anySub.asClientTopicCouple()); @@ -200,11 +200,11 @@ public class SubscriptionsStoreTest { @Test public void testMatchManyDeepSingle() { - Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE, false); + Subscription slashPlusSub = new Subscription("FAKE_CLI_ID_1", "/finance/+/ibm", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(slashPlusSub); store.add(slashPlusSub.asClientTopicCouple()); - Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_2", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE, false); + Subscription slashPlusDeepSub = new Subscription("FAKE_CLI_ID_2", "/+/stock/+", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(slashPlusDeepSub); store.add(slashPlusDeepSub.asClientTopicCouple()); @@ -214,7 +214,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti_allTheTree() { - Subscription sub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE, false); + Subscription sub = new Subscription("FAKE_CLI_ID_1", "#", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(sub); store.add(sub.asClientTopicCouple()); assertFalse(store.matches("finance").isEmpty()); @@ -224,7 +224,7 @@ public class SubscriptionsStoreTest { @Test public void testMatchSimpleMulti_zeroLevel() { //check MULTI in case of zero level match - Subscription sub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE, false); + Subscription sub = new Subscription("FAKE_CLI_ID_1", "finance/#", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(sub); store.add(sub.asClientTopicCouple()); assertFalse(store.matches("finance").isEmpty()); @@ -260,7 +260,7 @@ public class SubscriptionsStoreTest { memStore.initStore(); ISessionsStore aSessionsStore = memStore.sessionsStore(); store.init(aSessionsStore); - Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); + Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE); aSessionsStore.addNewSubscription(sub); store.add(sub.asClientTopicCouple()); assertFalse(store.matches(topic).isEmpty()); @@ -271,7 +271,7 @@ public class SubscriptionsStoreTest { MemoryStorageService memStore = new MemoryStorageService(); memStore.initStore(); store.init(memStore.sessionsStore()); - Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); + Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(sub); store.add(sub.asClientTopicCouple()); assertTrue(store.matches(topic).isEmpty()); @@ -281,7 +281,7 @@ public class SubscriptionsStoreTest { @Test public void testRemoveClientSubscriptions_existingClientID() { String cliendID = "FAKE_CLID_1"; - Subscription sub = new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false); + Subscription sub = new Subscription(cliendID, "finance/#", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(sub); store.add(sub.asClientTopicCouple()); @@ -295,7 +295,7 @@ public class SubscriptionsStoreTest { @Test public void testRemoveClientSubscriptions_notexistingClientID() { String clientID = "FAKE_CLID_1"; - Subscription s = new Subscription(clientID, "finance/#", AbstractMessage.QOSType.MOST_ONE, false); + Subscription s = new Subscription(clientID, "finance/#", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(s); store.add(s.asClientTopicCouple()); @@ -308,10 +308,10 @@ public class SubscriptionsStoreTest { @Test public void testOverlappingSubscriptions() { - Subscription genericSub = new Subscription("FAKE_CLI_ID_1", "a/+", AbstractMessage.QOSType.EXACTLY_ONCE, false); + Subscription genericSub = new Subscription("FAKE_CLI_ID_1", "a/+", AbstractMessage.QOSType.EXACTLY_ONCE); sessionsStore.addNewSubscription(genericSub); store.add(genericSub.asClientTopicCouple()); - Subscription specificSub = new Subscription("FAKE_CLI_ID_1", "a/b", AbstractMessage.QOSType.LEAST_ONE, false); + Subscription specificSub = new Subscription("FAKE_CLI_ID_1", "a/b", AbstractMessage.QOSType.LEAST_ONE); sessionsStore.addNewSubscription(specificSub); store.add(specificSub.asClientTopicCouple()); @@ -380,12 +380,12 @@ public class SubscriptionsStoreTest { aStore.init(sessionsStore); //subscribe a not active clientID1 to /topic sessionsStore.createNewSession("FAKE_CLI_ID_1", true).deactivate(); - Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); + Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(slashSub); aStore.add(slashSub.asClientTopicCouple()); //subscribe an active clientID2 to /topic - Subscription slashSub2 = new Subscription("FAKE_CLI_ID_2", "/topic", AbstractMessage.QOSType.MOST_ONE, false); + Subscription slashSub2 = new Subscription("FAKE_CLI_ID_2", "/topic", AbstractMessage.QOSType.MOST_ONE); sessionsStore.addNewSubscription(slashSub2); aStore.add(new ClientTopicCouple("FAKE_CLI_ID_2", "/topic")); @@ -401,10 +401,10 @@ public class SubscriptionsStoreTest { @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, false); + 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, false); + Subscription overrindingSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.EXACTLY_ONCE); session.subscribe("/topic", overrindingSubscription); store.add(overrindingSubscription.asClientTopicCouple()); @@ -421,15 +421,15 @@ public class SubscriptionsStoreTest { @Test public void duplicatedSubscriptionsWithDifferentQos() { ClientSession session2 = sessionsStore.createNewSession("client2", true); - Subscription client2Sub = new Subscription("client2", "client/test/b", AbstractMessage.QOSType.MOST_ONE, true); + Subscription client2Sub = new Subscription("client2", "client/test/b", AbstractMessage.QOSType.MOST_ONE); session2.subscribe("client/test/b", client2Sub); store.add(client2Sub.asClientTopicCouple()); ClientSession session1 = sessionsStore.createNewSession("client1", true); - Subscription client1SubQoS0 = new Subscription("client1", "client/test/b", AbstractMessage.QOSType.MOST_ONE, true); + Subscription client1SubQoS0 = new Subscription("client1", "client/test/b", AbstractMessage.QOSType.MOST_ONE); session1.subscribe("client/test/b", client1SubQoS0); store.add(client1SubQoS0.asClientTopicCouple()); - Subscription client1SubQoS2 = new Subscription("client1", "client/test/b", AbstractMessage.QOSType.EXACTLY_ONCE, true); + Subscription client1SubQoS2 = new Subscription("client1", "client/test/b", AbstractMessage.QOSType.EXACTLY_ONCE); session1.subscribe("client/test/b", client1SubQoS2); store.add(client1SubQoS2.asClientTopicCouple()); 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 877c4e1f..9121820d 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -70,9 +70,9 @@ 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, false); + 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, false); + Subscription overrindingSubscription = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.EXACTLY_ONCE); session1.subscribe("/topic", overrindingSubscription); //Verify