From 90cfe1ac2328fb967c6282d71d54836b4f48f219 Mon Sep 17 00:00:00 2001 From: "selva.andre" Date: Mon, 30 Jun 2014 22:46:11 +0200 Subject: [PATCH] Extracted the persistence dependency from SubscriptionsStore --- .../messaging/spi/impl/ProtocolProcessor.java | 4 ++++ .../messaging/spi/impl/SimpleMessaging.java | 10 +++++++--- .../impl/subscriptions/SubscriptionsStore.java | 18 ++++++------------ .../spi/impl/ProtocolProcessorTest.java | 9 +++++---- .../subscriptions/SubscriptionsStoreTest.java | 9 +++++---- 5 files changed, 27 insertions(+), 23 deletions(-) diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java index 9d375f46..81db4d18 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java @@ -245,6 +245,8 @@ class ProtocolProcessor implements EventHandler { private void processRemoveAllSubscriptions(String clientID) { LOG.info("cleaning old saved subscriptions for client <{}>", clientID); + //remove from log all subscriptions + m_storageService.removeAllSubscriptions(clientID); subscriptions.removeForClient(clientID); //remove also the messages stored of type QoS1/2 @@ -534,6 +536,8 @@ class ProtocolProcessor implements EventHandler { LOG.info("<{}> subscribed to topic <{}> with QoS {}", newSubscription.getClientId(), topic, AbstractMessage.QOSType.formatQoS(newSubscription.getRequestedQos())); + String clientID = newSubscription.getClientId(); + m_storageService.addNewSubscription(newSubscription, clientID); subscriptions.add(newSubscription); //scans retained messages to be published to the new subscription diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java index cdfbcf6b..4ec72bef 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java @@ -19,6 +19,7 @@ import com.lmax.disruptor.BatchEventProcessor; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.SequenceBarrier; +import java.util.List; import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -27,6 +28,7 @@ import java.util.concurrent.TimeUnit; import org.dna.mqtt.moquette.messaging.spi.IMessaging; import org.dna.mqtt.moquette.messaging.spi.IStorageService; import org.dna.mqtt.moquette.messaging.spi.impl.events.*; +import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.Subscription; import org.dna.mqtt.moquette.messaging.spi.impl.subscriptions.SubscriptionsStore; import org.dna.mqtt.moquette.proto.messages.PubCompMessage; import org.dna.mqtt.moquette.proto.messages.*; @@ -40,7 +42,9 @@ import org.slf4j.LoggerFactory; * * Singleton class that orchestrate the execution of the protocol. * - * Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion. + * Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion; + * the requests income from front Netty connectors and are dispatched to the + * ProtocolProcessor. * * @author andrea */ @@ -199,8 +203,8 @@ public class SimpleMessaging implements IMessaging, EventHandler { private void processInit(Properties props) { m_storageService = new HawtDBStorageService(); m_storageService.initStore(); - - subscriptions.init(m_storageService); + List storedSubscriptions = m_storageService.retrieveAllSubscriptions(); + subscriptions.init(storedSubscriptions); String passwdPath = props.getProperty("password_file"); String configPath = System.getProperty("moquette.path", null); diff --git a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java index d9534390..3462c10a 100644 --- a/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/subscriptions/SubscriptionsStore.java @@ -21,7 +21,6 @@ import java.util.Collections; import java.util.List; import java.util.Queue; import java.util.concurrent.LinkedBlockingDeque; -import org.dna.mqtt.moquette.messaging.spi.IPersistentSubscriptionStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -73,23 +72,18 @@ public class SubscriptionsStore { private TreeNode subscriptions = new TreeNode(null); private static final Logger LOG = LoggerFactory.getLogger(SubscriptionsStore.class); - private IPersistentSubscriptionStore m_storageService; - /** - * Initialize basic store structures, like the FS storage to maintain - * client's topics subscriptions + * Initialize the subscription tree with the list of subscriptions. */ - public void init(IPersistentSubscriptionStore storageService) { + public void init(List subscriptions) { LOG.debug("init invoked"); - m_storageService = storageService; - //reload any subscriptions persisted if (LOG.isDebugEnabled()) { LOG.debug("Reloading all stored subscriptions...subscription tree before {}", dumpTree()); } - for (Subscription subscription : m_storageService.retrieveAllSubscriptions()) { + for (Subscription subscription : subscriptions) { LOG.debug("Re-subscribing {} to topic {}", subscription.getClientId(), subscription.getTopicFilter()); addDirect(subscription); } @@ -135,8 +129,8 @@ public class SubscriptionsStore { addDirect(newSubscription); //log the subscription - String clientID = newSubscription.getClientId(); - m_storageService.addNewSubscription(newSubscription, clientID); +// String clientID = newSubscription.getClientId(); +// m_storageService.addNewSubscription(newSubscription, clientID); } @@ -177,7 +171,7 @@ public class SubscriptionsStore { subscriptions.removeClientSubscriptions(clientID); //remove from log all subscriptions - m_storageService.removeAllSubscriptions(clientID); +// m_storageService.removeAllSubscriptions(clientID); } public void deactivate(String clientID) { diff --git a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java index e604bff3..0a2a91b0 100644 --- a/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessorTest.java @@ -17,6 +17,7 @@ package org.dna.mqtt.moquette.messaging.spi.impl; import java.nio.ByteBuffer; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -152,7 +153,7 @@ public class ProtocolProcessorTest { m_mockAuthenticator = new MockAuthenticator(users); subscriptions = new SubscriptionsStore(); - subscriptions.init(m_storageService); + subscriptions.init(Collections.emptyList()); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_storageService, m_mockAuthenticator); } @@ -261,7 +262,7 @@ public class ProtocolProcessorTest { }; //simulate a connect that register a clientID to an IoSession - subs.init(m_storageService); + subs.init(Collections.emptyList()); m_processor.init(subs, m_storageService, null); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProcotolVersion((byte)3); @@ -300,7 +301,7 @@ public class ProtocolProcessorTest { }; //simulate a connect that register a clientID to an IoSession - subs.init(m_storageService); + subs.init(Collections.emptyList()); m_processor.init(subs, m_storageService, null); MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); @@ -408,7 +409,7 @@ public class ProtocolProcessorTest { } } }; - subs.init(new MemoryStorageService()); + subs.init(Collections.emptyList()); //simulate a connect that register a clientID to an IoSession m_processor.init(subs, m_storageService, null); 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 b72db12f..71201e21 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 @@ -19,6 +19,7 @@ import java.io.File; import java.io.IOException; import java.text.ParseException; import java.util.Arrays; +import java.util.Collections; import java.util.List; import org.dna.mqtt.moquette.messaging.spi.impl.MemoryStorageService; @@ -55,7 +56,7 @@ public class SubscriptionsStoreTest { pageFactory.open(); PageFile pageFile = pageFactory.getPageFile(); MultiIndexFactory multiIndexFactory = new MultiIndexFactory(pageFile); - store.init(new MemoryStorageService()); + store.init(Collections.emptyList()); } @After @@ -252,7 +253,7 @@ public class SubscriptionsStoreTest { private void assertMatch(String subscription, String topic) { store = new SubscriptionsStore(); - store.init(new MemoryStorageService()); + store.init(Collections.emptyList()); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); store.add(sub); assertFalse(store.matches(topic).isEmpty()); @@ -260,7 +261,7 @@ public class SubscriptionsStoreTest { private void assertNotMatch(String subscription, String topic) { store = new SubscriptionsStore(); - store.init(new MemoryStorageService()); + store.init(Collections.emptyList()); Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, AbstractMessage.QOSType.MOST_ONE, false); store.add(sub); assertTrue(store.matches(topic).isEmpty()); @@ -345,7 +346,7 @@ public class SubscriptionsStoreTest { @Test public void removeSubscription_withDifferentClients_subscribedSameTopic() { SubscriptionsStore aStore = new SubscriptionsStore(); - aStore.init(new MemoryStorageService()); + aStore.init(Collections.emptyList()); //subscribe a not active clientID1 to /topic Subscription slashSub = new Subscription("FAKE_CLI_ID_1", "/topic", AbstractMessage.QOSType.MOST_ONE, false); aStore.add(slashSub);