From a36d926460f8989baff7d252e47b3fcb9da66ae0 Mon Sep 17 00:00:00 2001 From: andsel Date: Tue, 13 Nov 2018 22:50:02 +0100 Subject: [PATCH] Implemented H2 based SubscriptionsRepository --- broker/build.gradle | 5 +- .../broker/ISubscriptionsRepository.java | 2 + .../main/java/io/moquette/broker/Server.java | 17 ++++--- .../CTrieSubscriptionDirectory.java | 25 ++++++--- .../broker/subscriptions/Subscription.java | 27 +--------- .../io/moquette/persistence/H2Builder.java | 51 +++++++++++++++++++ .../H2SubscriptionsRepository.java | 48 +++++++++++++++++ .../persistence/MemoryStorageService.java | 1 + .../MemorySubscriptionsRepository.java | 11 +++- .../src/main/java/io/moquette/spi/IStore.java | 1 + .../broker/MQTTConnectionConnectTest.java | 1 + .../broker/MQTTConnectionPublishTest.java | 1 + .../broker/PostOfficeInternalPublishTest.java | 1 + .../broker/PostOfficePublishTest.java | 1 + .../broker/PostOfficeSubscribeTest.java | 1 + .../broker/PostOfficeUnsubscribeTest.java | 1 + .../moquette/broker/SessionRegistryTest.java | 1 + settings.gradle | 2 - 18 files changed, 151 insertions(+), 46 deletions(-) create mode 100644 broker/src/main/java/io/moquette/persistence/H2Builder.java create mode 100644 broker/src/main/java/io/moquette/persistence/H2SubscriptionsRepository.java rename broker/src/main/java/io/moquette/{broker => persistence}/MemorySubscriptionsRepository.java (75%) diff --git a/broker/build.gradle b/broker/build.gradle index cdd0e11e..4a0d96ea 100644 --- a/broker/build.gradle +++ b/broker/build.gradle @@ -13,6 +13,8 @@ dependencies { compile group: 'io.netty', name: 'netty-transport-native-epoll', version: nettyVersion, classifier: 'linux-x86_64' // ${os.detected.name}-${os.detected.arch} + compile group: 'com.h2database', name: 'h2-mvstore', version: '1.4.192' + compile group: 'com.zaxxer', name: 'HikariCP', version:'2.4.7' compile group: 'io.dropwizard.metrics', name: 'metrics-core', version:'3.2.2' @@ -26,12 +28,11 @@ dependencies { compile group: 'org.slf4j', name: 'slf4j-api', version:'1.7.5' runtime group: 'org.slf4j', name: 'slf4j-log4j12', version:'1.7.5' // only for local tests - testCompile group: 'org.slf4j', name: 'slf4j-log4j12', version:'1.7.5' + testCompile group: 'org.slf4j', name: 'slf4j-log4j12', version:'1.7.5' testCompile group: 'org.fusesource.mqtt-client', name: 'mqtt-client', version:'1.12' testCompile group: 'org.eclipse.paho', name: 'org.eclipse.paho.client.mqttv3', version:'1.2.0' testCompile group: 'org.eclipse.jetty.websocket', name: 'websocket-client', version:'9.2.0.M1' - testCompile group: 'com.h2database', name: 'h2', version:'1.4.191' testRuntime group: 'io.netty', name: 'netty-tcnative', version: '2.0.10.Final', classifier: 'linux-x86_64' } diff --git a/broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java b/broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java index 5b86c8be..bf7c1d66 100644 --- a/broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java +++ b/broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java @@ -24,4 +24,6 @@ public interface ISubscriptionsRepository { List listAllSubscriptions(); void addNewSubscription(Subscription subscription); + + void removeSubscription(String topic, String clientID); } diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java index 4391361d..aba97488 100644 --- a/broker/src/main/java/io/moquette/broker/Server.java +++ b/broker/src/main/java/io/moquette/broker/Server.java @@ -18,8 +18,8 @@ package io.moquette.broker; import io.moquette.BrokerConstants; import io.moquette.broker.config.*; import io.moquette.interception.InterceptHandler; -import io.moquette.persistence.MemoryStorageService; -import io.moquette.spi.ISessionsStore; +import io.moquette.persistence.H2Builder; +import io.moquette.persistence.MemorySubscriptionsRepository; import io.moquette.interception.BrokerInterceptor; import io.moquette.broker.security.*; import io.moquette.broker.subscriptions.CTrieSubscriptionDirectory; @@ -159,11 +159,14 @@ public class Server { authenticator = initializeAuthenticator(authenticator, config); authorizatorPolicy = initializeAuthorizatorPolicy(authorizatorPolicy, config); - // TODO user real implementation DBG - MemoryStorageService memStorage = new MemoryStorageService(null, null); - ISessionsStore sessionStore = memStorage.sessionsStore(); - ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository(); - // DBG + final ISubscriptionsRepository subscriptionsRepository; + if (persistencePath != null && !persistencePath.isEmpty()) { + LOG.trace("Configuring H2 subscriptions store to {}", persistencePath); + subscriptionsRepository = new H2Builder(config, scheduler).initStore().subscriptionsRepository(); + } else { + LOG.trace("Configuring in-memory subscriptions store"); + subscriptionsRepository = new MemorySubscriptionsRepository(); + } ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); subscriptions.init(subscriptionsRepository); diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java b/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java index 0a345ab3..72874dc6 100644 --- a/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java +++ b/broker/src/main/java/io/moquette/broker/subscriptions/CTrieSubscriptionDirectory.java @@ -28,8 +28,8 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { private static final Token ROOT = new Token("root"); private static final INode NO_PARENT = null; - INode root; - private volatile ISubscriptionsRepository sessionsRepository; + private INode root; + private volatile ISubscriptionsRepository subscriptionsRepository; interface IVisitor { @@ -43,23 +43,22 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { } @Override - public void init(ISubscriptionsRepository sessionsRepository) { + public void init(ISubscriptionsRepository subscriptionsRepository) { LOG.info("Initializing CTrie"); final CNode mainNode = new CNode(); mainNode.token = ROOT; this.root = new INode(mainNode); LOG.info("Initializing subscriptions store..."); - this.sessionsRepository = sessionsRepository; + this.subscriptionsRepository = subscriptionsRepository; // reload any subscriptions persisted if (LOG.isTraceEnabled()) { LOG.trace("Reloading all stored subscriptions. SubscriptionTree = {}", dumpTree()); } - for (Subscription subscription : this.sessionsRepository.listAllSubscriptions()) { - LOG.info("Re-subscribing client to topic CId={}, topicFilter={}", subscription.clientId, - subscription.topicFilter); - add(subscription); + for (Subscription subscription : this.subscriptionsRepository.listAllSubscriptions()) { + LOG.debug("Re-subscribing {}", subscription); + addToTree(subscription); } if (LOG.isTraceEnabled()) { LOG.trace("Stored subscriptions have been reloaded. SubscriptionTree = {}", dumpTree()); @@ -160,6 +159,11 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { @Override public void add(Subscription newSubscription) { + addToTree(newSubscription); + subscriptionsRepository.addNewSubscription(newSubscription); + } + + private void addToTree(Subscription newSubscription) { Action res; do { res = insert(newSubscription.topicFilter, this.root, newSubscription); @@ -230,6 +234,11 @@ public class CTrieSubscriptionDirectory implements ISubscriptionsDirectory { */ @Override public void removeSubscription(Topic topic, String clientID) { + removeFromTree(topic, clientID); + this.subscriptionsRepository.removeSubscription(topic.toString(), clientID); + } + + private void removeFromTree(Topic topic, String clientID) { Action res; do { res = remove(clientID, topic, this.root, NO_PARENT); diff --git a/broker/src/main/java/io/moquette/broker/subscriptions/Subscription.java b/broker/src/main/java/io/moquette/broker/subscriptions/Subscription.java index 4a42a1e5..54885a83 100644 --- a/broker/src/main/java/io/moquette/broker/subscriptions/Subscription.java +++ b/broker/src/main/java/io/moquette/broker/subscriptions/Subscription.java @@ -29,33 +29,17 @@ public final class Subscription implements Serializable { private final MqttQoS requestedQos; // max QoS acceptable final String clientId; final Topic topicFilter; - private final boolean active; public Subscription(String clientId, Topic topicFilter, MqttQoS requestedQos) { this.requestedQos = requestedQos; this.clientId = clientId; this.topicFilter = topicFilter; - this.active = true; } public Subscription(Subscription orig) { this.requestedQos = orig.requestedQos; this.clientId = orig.clientId; this.topicFilter = orig.topicFilter; - this.active = orig.active; - } - - /** - * Constructor with undefined maximum QoS - * - * @param clientId id of client owning this subscription. - * @param topicFilter the topic to subscribe to. - * */ - public Subscription(String clientId, Topic topicFilter) { - this.requestedQos = null; - this.clientId = clientId; - this.topicFilter = topicFilter; - this.active = true; } public String getClientId() { @@ -70,10 +54,6 @@ public final class Subscription implements Serializable { return topicFilter; } - public boolean isActive() { - return active; - } - public boolean qosLessThan(Subscription sub) { return requestedQos.value() < sub.requestedQos.value(); } @@ -101,12 +81,7 @@ public final class Subscription implements Serializable { @Override public String toString() { - return String.format( - "[filter:%s, cliID: %s, qos: %s, active: %s]", - this.topicFilter, - this.clientId, - this.requestedQos, - this.active); + return String.format("[filter:%s, clientID: %s, qos: %s]", topicFilter, clientId, requestedQos); } @Override diff --git a/broker/src/main/java/io/moquette/persistence/H2Builder.java b/broker/src/main/java/io/moquette/persistence/H2Builder.java new file mode 100644 index 00000000..a70e5d6e --- /dev/null +++ b/broker/src/main/java/io/moquette/persistence/H2Builder.java @@ -0,0 +1,51 @@ +package io.moquette.persistence; + +import io.moquette.BrokerConstants; +import io.moquette.broker.ISubscriptionsRepository; +import io.moquette.broker.config.IConfig; +import org.h2.mvstore.MVStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +public class H2Builder { + + private static final Logger LOG = LoggerFactory.getLogger(H2Builder.class); + + private final String storePath; + private final int autosaveInterval; // in seconds + private final ScheduledExecutorService scheduler; + private MVStore mvStore; + + public H2Builder(IConfig props, ScheduledExecutorService scheduler) { + this.storePath = props.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, ""); + final String autosaveProp = props.getProperty(BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME, "30"); + this.autosaveInterval = Integer.parseInt(autosaveProp); + this.scheduler = scheduler; + } + + @SuppressWarnings("FutureReturnValueIgnored") + public H2Builder initStore() { + LOG.info("Initializing H2 store"); + if (storePath == null || storePath.isEmpty()) { + throw new IllegalArgumentException("H2 store path can't be null or empty"); + } + mvStore = new MVStore.Builder() + .fileName(storePath) + .autoCommitDisabled() + .open(); + + LOG.trace("Scheduling H2 commit task"); + scheduler.scheduleWithFixedDelay(() -> { + LOG.trace("Committing to H2"); + mvStore.commit(); + }, autosaveInterval, autosaveInterval, TimeUnit.SECONDS); + return this; + } + + public ISubscriptionsRepository subscriptionsRepository() { + return new H2SubscriptionsRepository(mvStore); + } +} diff --git a/broker/src/main/java/io/moquette/persistence/H2SubscriptionsRepository.java b/broker/src/main/java/io/moquette/persistence/H2SubscriptionsRepository.java new file mode 100644 index 00000000..e733ebfc --- /dev/null +++ b/broker/src/main/java/io/moquette/persistence/H2SubscriptionsRepository.java @@ -0,0 +1,48 @@ +package io.moquette.persistence; + +import io.moquette.broker.ISubscriptionsRepository; +import io.moquette.broker.subscriptions.Subscription; +import org.h2.mvstore.Cursor; +import org.h2.mvstore.MVMap; +import org.h2.mvstore.MVStore; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.List; + +public class H2SubscriptionsRepository implements ISubscriptionsRepository { + + private static final Logger LOG = LoggerFactory.getLogger(H2SubscriptionsRepository.class); + private static final String SUBSCRIPTIONS_MAP = "subscriptions"; + + private MVMap subscriptions; + + H2SubscriptionsRepository(MVStore mvStore) { + this.subscriptions = mvStore.openMap(SUBSCRIPTIONS_MAP); + } + + @Override + public List listAllSubscriptions() { + LOG.debug("Retrieving existing subscriptions"); + + List results = new ArrayList<>(); + Cursor mapCursor = subscriptions.cursor(null); + while (mapCursor.hasNext()) { + String subscriptionStr = mapCursor.next(); + results.add(mapCursor.getValue()); + } + LOG.debug("Loaded {} subscriptions", results.size()); + return results; + } + + @Override + public void addNewSubscription(Subscription subscription) { + subscriptions.put(subscription.getTopicFilter() + "-" + subscription.getClientId(), subscription); + } + + @Override + public void removeSubscription(String topic, String clientID) { + + } +} diff --git a/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java b/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java index 9bf87557..3aea9de4 100644 --- a/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java @@ -23,6 +23,7 @@ import io.moquette.spi.IStore; import java.util.concurrent.ScheduledExecutorService; +@Deprecated public class MemoryStorageService implements IStore { private MemorySessionStore m_sessionsStore; diff --git a/broker/src/main/java/io/moquette/broker/MemorySubscriptionsRepository.java b/broker/src/main/java/io/moquette/persistence/MemorySubscriptionsRepository.java similarity index 75% rename from broker/src/main/java/io/moquette/broker/MemorySubscriptionsRepository.java rename to broker/src/main/java/io/moquette/persistence/MemorySubscriptionsRepository.java index 6003524e..04472dd2 100644 --- a/broker/src/main/java/io/moquette/broker/MemorySubscriptionsRepository.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySubscriptionsRepository.java @@ -13,8 +13,9 @@ * * You may elect to redistribute this code under either of these licenses. */ -package io.moquette.broker; +package io.moquette.persistence; +import io.moquette.broker.ISubscriptionsRepository; import io.moquette.broker.subscriptions.Subscription; import java.util.ArrayList; @@ -34,4 +35,12 @@ public class MemorySubscriptionsRepository implements ISubscriptionsRepository { public void addNewSubscription(Subscription subscription) { subscriptions.add(subscription); } + + @Override + public void removeSubscription(String topic, String clientID) { + subscriptions.stream() + .filter(s -> s.getTopicFilter().toString().equals(topic) && s.getClientId().equals(clientID)) + .findFirst() + .ifPresent(subscriptions::remove); + } } diff --git a/broker/src/main/java/io/moquette/spi/IStore.java b/broker/src/main/java/io/moquette/spi/IStore.java index 8ce4424a..8af6e71c 100644 --- a/broker/src/main/java/io/moquette/spi/IStore.java +++ b/broker/src/main/java/io/moquette/spi/IStore.java @@ -16,6 +16,7 @@ package io.moquette.spi; +@Deprecated public interface IStore { void initStore(); diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java index c69d155d..45573d7b 100644 --- a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java +++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java @@ -19,6 +19,7 @@ import io.moquette.broker.security.PermitAllAuthorizatorPolicy; import io.moquette.broker.subscriptions.CTrieSubscriptionDirectory; import io.moquette.broker.subscriptions.ISubscriptionsDirectory; import io.moquette.broker.security.IAuthenticator; +import io.moquette.persistence.MemorySubscriptionsRepository; import io.netty.channel.Channel; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.mqtt.MqttConnectMessage; diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java index 27ab581d..967e2e01 100644 --- a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java +++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java @@ -19,6 +19,7 @@ import io.moquette.broker.security.PermitAllAuthorizatorPolicy; import io.moquette.broker.subscriptions.CTrieSubscriptionDirectory; import io.moquette.broker.subscriptions.ISubscriptionsDirectory; import io.moquette.broker.security.IAuthenticator; +import io.moquette.persistence.MemorySubscriptionsRepository; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.embedded.EmbeddedChannel; diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java index 3858cb0d..d1cd900c 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java @@ -20,6 +20,7 @@ import io.moquette.broker.subscriptions.CTrieSubscriptionDirectory; import io.moquette.broker.subscriptions.ISubscriptionsDirectory; import io.moquette.broker.subscriptions.Subscription; import io.moquette.broker.subscriptions.Topic; +import io.moquette.persistence.MemorySubscriptionsRepository; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.embedded.EmbeddedChannel; diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java index 3660c52a..c2abe8ed 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java @@ -20,6 +20,7 @@ import io.moquette.broker.subscriptions.CTrieSubscriptionDirectory; import io.moquette.broker.subscriptions.ISubscriptionsDirectory; import io.moquette.broker.subscriptions.Subscription; import io.moquette.broker.subscriptions.Topic; +import io.moquette.persistence.MemorySubscriptionsRepository; import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBufUtil; import io.netty.buffer.Unpooled; diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java index 7fa0ce63..e5e44570 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java @@ -22,6 +22,7 @@ import io.moquette.broker.subscriptions.Subscription; import io.moquette.broker.subscriptions.Topic; import io.moquette.broker.security.IAuthenticator; import io.moquette.broker.security.IAuthorizatorPolicy; +import io.moquette.persistence.MemorySubscriptionsRepository; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java index 548e188d..05050450 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java @@ -21,6 +21,7 @@ import io.moquette.broker.subscriptions.ISubscriptionsDirectory; import io.moquette.broker.subscriptions.Subscription; import io.moquette.broker.subscriptions.Topic; import io.moquette.broker.security.IAuthenticator; +import io.moquette.persistence.MemorySubscriptionsRepository; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java index 2ecf62e6..626b505d 100644 --- a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java +++ b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java @@ -19,6 +19,7 @@ import io.moquette.broker.security.PermitAllAuthorizatorPolicy; import io.moquette.broker.subscriptions.CTrieSubscriptionDirectory; import io.moquette.broker.subscriptions.ISubscriptionsDirectory; import io.moquette.broker.security.IAuthenticator; +import io.moquette.persistence.MemorySubscriptionsRepository; import io.netty.channel.Channel; import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.mqtt.MqttConnectMessage; diff --git a/settings.gradle b/settings.gradle index f822d971..4e16aab9 100644 --- a/settings.gradle +++ b/settings.gradle @@ -2,11 +2,9 @@ rootProject.name = 'moquette-parent' include ':moquette-broker' include ':moquette-distribution' include ':moquette-h2-storage' -include ':moquette-mapdb-storage' include ':moquette-embedded-test' project(':moquette-broker').projectDir = "$rootDir/broker" as File project(':moquette-distribution').projectDir = "$rootDir/distribution" as File project(':moquette-embedded-test').projectDir = "$rootDir/embedding_moquette" as File project(':moquette-h2-storage').projectDir = "$rootDir/h2_storage" as File -project(':moquette-mapdb-storage').projectDir = "$rootDir/mapdb_storage" as File