diff --git a/broker/build.gradle b/broker/build.gradle index 517facfd..6c8ba3c8 100644 --- a/broker/build.gradle +++ b/broker/build.gradle @@ -33,7 +33,7 @@ dependencies { 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' + testCompile group: 'com.h2database', name: 'h2', version:'1.4.192' 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/IQueueRepository.java b/broker/src/main/java/io/moquette/broker/IQueueRepository.java new file mode 100644 index 00000000..602949f2 --- /dev/null +++ b/broker/src/main/java/io/moquette/broker/IQueueRepository.java @@ -0,0 +1,8 @@ +package io.moquette.broker; + +import java.util.Queue; + +public interface IQueueRepository { + + Queue createQueue(String cli, boolean clean); +} diff --git a/broker/src/main/java/io/moquette/broker/MemoryQueueRepository.java b/broker/src/main/java/io/moquette/broker/MemoryQueueRepository.java new file mode 100644 index 00000000..1b840529 --- /dev/null +++ b/broker/src/main/java/io/moquette/broker/MemoryQueueRepository.java @@ -0,0 +1,12 @@ +package io.moquette.broker; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +public class MemoryQueueRepository implements IQueueRepository { + + @Override + public Queue createQueue(String cli, boolean clean) { + return new ConcurrentLinkedQueue<>(); + } +} diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java index 4c5a6e46..4ae6c828 100644 --- a/broker/src/main/java/io/moquette/broker/Server.java +++ b/broker/src/main/java/io/moquette/broker/Server.java @@ -161,18 +161,21 @@ public class Server { authorizatorPolicy = initializeAuthorizatorPolicy(authorizatorPolicy, config); final ISubscriptionsRepository subscriptionsRepository; + final IQueueRepository queueRepository; if (persistencePath != null && !persistencePath.isEmpty()) { LOG.trace("Configuring H2 subscriptions store to {}", persistencePath); - h2Builder = new H2Builder(config, scheduler); - subscriptionsRepository = h2Builder.initStore().subscriptionsRepository(); + h2Builder = new H2Builder(config, scheduler).initStore(); + subscriptionsRepository = h2Builder.subscriptionsRepository(); + queueRepository = h2Builder.queueRepository(); } else { LOG.trace("Configuring in-memory subscriptions store"); subscriptionsRepository = new MemorySubscriptionsRepository(); + queueRepository = new MemoryQueueRepository(); } ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); subscriptions.init(subscriptionsRepository); - SessionRegistry sessions = new SessionRegistry(subscriptions, interceptor); + SessionRegistry sessions = new SessionRegistry(subscriptions, queueRepository); dispatcher = new PostOffice(subscriptions, authorizatorPolicy, new MemoryRetainedRepository(), sessions, interceptor); final BrokerConfiguration brokerConfig = new BrokerConfiguration(config); diff --git a/broker/src/main/java/io/moquette/broker/Session.java b/broker/src/main/java/io/moquette/broker/Session.java index ea82ef01..c1214398 100644 --- a/broker/src/main/java/io/moquette/broker/Session.java +++ b/broker/src/main/java/io/moquette/broker/Session.java @@ -97,10 +97,8 @@ class Session { private final AtomicInteger inflightSlots = new AtomicInteger(INFLIGHT_WINDOW_SIZE); // this should be configurable Session(String clientId, boolean clean, Will will, Queue sessionQueue) { - this.clientId = clientId; - this.clean = clean; + this(clean, clientId, sessionQueue); this.will = will; - this.sessionQueue = sessionQueue; } Session(boolean clean, String clientId, Queue sessionQueue) { diff --git a/broker/src/main/java/io/moquette/broker/SessionRegistry.java b/broker/src/main/java/io/moquette/broker/SessionRegistry.java index 47dbfe52..f2287d14 100644 --- a/broker/src/main/java/io/moquette/broker/SessionRegistry.java +++ b/broker/src/main/java/io/moquette/broker/SessionRegistry.java @@ -16,7 +16,6 @@ package io.moquette.broker; import io.moquette.broker.Session.SessionStatus; -import io.moquette.interception.BrokerInterceptor; import io.moquette.broker.subscriptions.ISubscriptionsDirectory; import io.moquette.broker.subscriptions.Subscription; import io.moquette.broker.subscriptions.Topic; @@ -32,9 +31,9 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; -class SessionRegistry { +public class SessionRegistry { - abstract static class EnqueuedMessage { + public abstract static class EnqueuedMessage { } static class PublishedMessage extends EnqueuedMessage { @@ -61,10 +60,12 @@ class SessionRegistry { private final ConcurrentMap pool = new ConcurrentHashMap<>(); private final ISubscriptionsDirectory subscriptionsDirectory; + private final IQueueRepository queueRepository; private final ConcurrentMap> queues = new ConcurrentHashMap<>(); - SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory, BrokerInterceptor m_interceptor) { + SessionRegistry(ISubscriptionsDirectory subscriptionsDirectory, IQueueRepository queueRepository) { this.subscriptionsDirectory = subscriptionsDirectory; + this.queueRepository = queueRepository; } void bindToSession(MQTTConnection mqttConnection, MqttConnectMessage msg, String clientId) { @@ -178,9 +179,9 @@ class SessionRegistry { } private Session createNewSession(MQTTConnection mqttConnection, MqttConnectMessage msg, String clientId) { - final Queue sessionQueue = - queues.computeIfAbsent(clientId, (String cli) -> new ConcurrentLinkedQueue<>()); final boolean clean = msg.variableHeader().isCleanSession(); + final Queue sessionQueue = + queues.computeIfAbsent(clientId, (String cli) -> queueRepository.createQueue(cli, clean)); final Session newSession; if (msg.variableHeader().isWillFlag()) { final Session.Will will = createWill(msg); diff --git a/broker/src/main/java/io/moquette/persistence/H2Builder.java b/broker/src/main/java/io/moquette/persistence/H2Builder.java index 324528c1..c82a0aeb 100644 --- a/broker/src/main/java/io/moquette/persistence/H2Builder.java +++ b/broker/src/main/java/io/moquette/persistence/H2Builder.java @@ -1,6 +1,7 @@ package io.moquette.persistence; import io.moquette.BrokerConstants; +import io.moquette.broker.IQueueRepository; import io.moquette.broker.ISubscriptionsRepository; import io.moquette.broker.config.IConfig; import org.h2.mvstore.MVStore; @@ -52,4 +53,8 @@ public class H2Builder { public void closeStore() { mvStore.close(); } + + public IQueueRepository queueRepository() { + return new H2QueueRepository(mvStore); + } } diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java b/broker/src/main/java/io/moquette/persistence/H2PersistentQueue.java similarity index 97% rename from h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java rename to broker/src/main/java/io/moquette/persistence/H2PersistentQueue.java index d260720a..d890beb4 100644 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2PersistentQueue.java +++ b/broker/src/main/java/io/moquette/persistence/H2PersistentQueue.java @@ -13,7 +13,7 @@ * * You may elect to redistribute this code under either of these licenses. */ -package io.moquette.persistence.h2; +package io.moquette.persistence; import org.h2.mvstore.MVMap; import org.h2.mvstore.MVStore; @@ -88,6 +88,7 @@ class H2PersistentQueue extends AbstractQueue { } final long nextTail = tail.getAndIncrement(); final T tail = this.queueMap.get(nextTail); + queueMap.remove(nextTail); this.metadataMap.put("tail", nextTail + 1); return tail; } diff --git a/broker/src/main/java/io/moquette/persistence/H2QueueRepository.java b/broker/src/main/java/io/moquette/persistence/H2QueueRepository.java new file mode 100644 index 00000000..84e940ca --- /dev/null +++ b/broker/src/main/java/io/moquette/persistence/H2QueueRepository.java @@ -0,0 +1,25 @@ +package io.moquette.persistence; + +import io.moquette.broker.IQueueRepository; +import io.moquette.broker.SessionRegistry; +import org.h2.mvstore.MVStore; + +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; + +public class H2QueueRepository implements IQueueRepository { + + private MVStore mvStore; + + public H2QueueRepository(MVStore mvStore) { + this.mvStore = mvStore; + } + + @Override + public Queue createQueue(String cli, boolean clean) { + if (!clean) { + return new H2PersistentQueue<>(mvStore, cli); + } + return new ConcurrentLinkedQueue<>(); + } +} diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java index 45573d7b..d5aa50b6 100644 --- a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java +++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java @@ -52,6 +52,7 @@ public class MQTTConnectionConnectTest { private static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false); private IAuthenticator mockAuthenticator; private PostOffice postOffice; + private MemoryQueueRepository queueRepository; @Before public void setUp() { @@ -62,8 +63,9 @@ public class MQTTConnectionConnectTest { ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository(); subscriptions.init(subscriptionsRepository); + queueRepository = new MemoryQueueRepository(); - sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + sessionRegistry = new SessionRegistry(subscriptions, queueRepository); postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java index 967e2e01..3db19c1b 100644 --- a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java +++ b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java @@ -45,6 +45,7 @@ public class MQTTConnectionPublishTest { private EmbeddedChannel channel; private SessionRegistry sessionRegistry; private MqttMessageBuilders.ConnectBuilder connMsg; + private MemoryQueueRepository queueRepository; @Before public void setUp() { @@ -67,8 +68,9 @@ public class MQTTConnectionPublishTest { ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository(); subscriptions.init(subscriptionsRepository); + queueRepository = new MemoryQueueRepository(); - sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + sessionRegistry = new SessionRegistry(subscriptions, queueRepository); final PostOffice postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java index d1cd900c..6b53ba1f 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java @@ -54,6 +54,7 @@ public class PostOfficeInternalPublishTest { private static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID = new BrokerConfiguration(true, true, false); private MemoryRetainedRepository retainedRepository; + private MemoryQueueRepository queueRepository; @Before public void setUp() { @@ -82,9 +83,9 @@ public class PostOfficeInternalPublishTest { ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository(); subscriptions.init(subscriptionsRepository); retainedRepository = new MemoryRetainedRepository(); + queueRepository = new MemoryQueueRepository(); - SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, - ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository); sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); return sessionRegistry; diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java index c2abe8ed..52aa9540 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java @@ -64,6 +64,7 @@ public class PostOfficePublishTest { static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID = new BrokerConfiguration(true, true, false); private MemoryRetainedRepository retainedRepository; + private MemoryQueueRepository queueRepository; @Before public void setUp() { @@ -89,9 +90,9 @@ public class PostOfficePublishTest { ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository(); subscriptions.init(subscriptionsRepository); retainedRepository = new MemoryRetainedRepository(); + queueRepository = new MemoryQueueRepository(); - SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, - ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + SessionRegistry sessionRegistry = new SessionRegistry(subscriptions, queueRepository); sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), retainedRepository, sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); return sessionRegistry; diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java index e5e44570..b91fb51e 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java @@ -65,6 +65,7 @@ public class PostOfficeSubscribeTest { private IAuthenticator mockAuthenticator; private SessionRegistry sessionRegistry; public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false); + private MemoryQueueRepository queueRepository; @Before public void setUp() { @@ -87,8 +88,9 @@ public class PostOfficeSubscribeTest { subscriptions = new CTrieSubscriptionDirectory(); ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository(); subscriptions.init(subscriptionsRepository); + queueRepository = new MemoryQueueRepository(); - sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + sessionRegistry = new SessionRegistry(subscriptions, queueRepository); sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); } diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java index 05050450..fea802bd 100644 --- a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java +++ b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java @@ -56,6 +56,7 @@ public class PostOfficeUnsubscribeTest { private IAuthenticator mockAuthenticator; private SessionRegistry sessionRegistry; public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false); + private MemoryQueueRepository queueRepository; @Before public void setUp() { @@ -78,8 +79,9 @@ public class PostOfficeUnsubscribeTest { subscriptions = new CTrieSubscriptionDirectory(); ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository(); subscriptions.init(subscriptionsRepository); + queueRepository = new MemoryQueueRepository(); - sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + sessionRegistry = new SessionRegistry(subscriptions, queueRepository); sut = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), sessionRegistry, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); } diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java index 626b505d..8bb5add5 100644 --- a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java +++ b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java @@ -47,6 +47,7 @@ public class SessionRegistryTest { private MqttMessageBuilders.ConnectBuilder connMsg; private static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZEROBYTE_CLIENT_ID = new BrokerConfiguration(true, true, false); + private MemoryQueueRepository queueRepository; @Before public void setUp() { @@ -67,8 +68,9 @@ public class SessionRegistryTest { ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory(); ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository(); subscriptions.init(subscriptionsRepository); + queueRepository = new MemoryQueueRepository(); - sut = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); + sut = new SessionRegistry(subscriptions, queueRepository); final PostOffice postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(), sut, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR); diff --git a/h2_storage/src/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java b/broker/src/test/java/io/moquette/persistence/H2PersistentQueueTest.java similarity index 98% rename from h2_storage/src/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java rename to broker/src/test/java/io/moquette/persistence/H2PersistentQueueTest.java index 2f68944e..43e3fb35 100644 --- a/h2_storage/src/test/java/io/moquette/persistence/h2/H2PersistentQueueTest.java +++ b/broker/src/test/java/io/moquette/persistence/H2PersistentQueueTest.java @@ -13,7 +13,7 @@ * * You may elect to redistribute this code under either of these licenses. */ -package io.moquette.persistence.h2; +package io.moquette.persistence; import io.moquette.BrokerConstants; import org.h2.mvstore.MVStore; diff --git a/distribution/build.gradle b/distribution/build.gradle index 594ba1ad..61be7546 100644 --- a/distribution/build.gradle +++ b/distribution/build.gradle @@ -8,8 +8,6 @@ mainClassName = "io.moquette.integration.Server" dependencies { compile project(':moquette-broker') - compile project(':moquette-h2-storage') - //compile project(':moquette-mapdb-storage') compile group: 'org.slf4j', name: 'slf4j-log4j12', version:'1.7.5' } diff --git a/h2_storage/build.gradle b/h2_storage/build.gradle deleted file mode 100644 index cbe04fc6..00000000 --- a/h2_storage/build.gradle +++ /dev/null @@ -1,9 +0,0 @@ - -description = 'Moquette - H2 storage' - -dependencies { - compile project(':moquette-broker') - compile group: 'com.h2database', name: 'h2-mvstore', version: '1.4.192' - - testCompile project(path:':moquette-broker', configuration: 'archives') -} diff --git a/h2_storage/src/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java b/h2_storage/src/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java deleted file mode 100644 index 2caf450e..00000000 --- a/h2_storage/src/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java +++ /dev/null @@ -1,58 +0,0 @@ -/* - * Copyright (c) 2012-2018 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ -package io.moquette.persistence.h2; - -import static org.junit.Assert.assertFalse; - -public class H2MessagesStoreTest /*extends MessageStoreTCK*/ { - -// H2PersistentStore storageService; -// -// private ScheduledExecutorService scheduler; -// -// @Before -// public void setUp() throws Exception { -// scheduler = Executors.newScheduledThreadPool(1); -// -// cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); -// Properties props = new Properties(); -// props.setProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, BrokerConstants.DEFAULT_PERSISTENT_PATH); -// IConfig conf = new MemoryConfig(props); -// storageService = new H2PersistentStore(conf, scheduler); -// storageService.initStore(); -// messagesStore = storageService.messagesStore(); -// sessionsStore = storageService.sessionsStore(); -// this.sessionsRepository = new SessionsRepository(sessionsStore, null); -// } -// -// @After -// public void tearDown() { -// if (storageService != null) { -// storageService.close(); -// } -// -// scheduler.shutdown(); -// cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); -// } -// -// public static void cleanPersistenceFile(String fileName) { -// File dbFile = new File(fileName); -// if (dbFile.exists()) { -// dbFile.delete(); -// } -// assertFalse(dbFile.exists()); -// } -} diff --git a/h2_storage/src/test/java/io/moquette/persistence/h2/H2SessionsStoreTest.java b/h2_storage/src/test/java/io/moquette/persistence/h2/H2SessionsStoreTest.java deleted file mode 100644 index eddf9aa0..00000000 --- a/h2_storage/src/test/java/io/moquette/persistence/h2/H2SessionsStoreTest.java +++ /dev/null @@ -1,119 +0,0 @@ -/* - * Copyright (c) 2012-2018 The original author or authors - * ------------------------------------------------------ - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * and Apache License v2.0 which accompanies this distribution. - * - * The Eclipse Public License is available at - * http://www.eclipse.org/legal/epl-v10.html - * - * The Apache License v2.0 is available at - * http://www.opensource.org/licenses/apache2.0.php - * - * You may elect to redistribute this code under either of these licenses. - */ -package io.moquette.persistence.h2; - -//import static io.moquette.persistence.h2.H2MessagesStoreTest.cleanPersistenceFile; -import static java.time.LocalDateTime.now; -import static org.assertj.core.api.Assertions.assertThat; - -public class H2SessionsStoreTest { - -// private static final Logger LOG = LoggerFactory.getLogger(H2SessionsStoreTest.class); -// -// private H2PersistentStore storageService; -// private ScheduledExecutorService scheduler; -// -// @Before -// public void setUp() throws Exception { -// scheduler = Executors.newScheduledThreadPool(1); -// -// cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); -// Properties props = new Properties(); -// props.setProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, BrokerConstants.DEFAULT_PERSISTENT_PATH); -// IConfig conf = new MemoryConfig(props); -// storageService = new H2PersistentStore(conf, scheduler); -// storageService.initStore(); -// } -// -// @After -// public void tearDown() { -// if (storageService != null) { -// storageService.close(); -// } -// -// scheduler.shutdown(); -// cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); -// } -// -// @Test -// public void testQueryByExpirationDate() { -// H2SessionsStore sessionsStore = (H2SessionsStore) storageService.sessionsStore(); -// LocalDateTime sessionCreationTime = LocalDateTime.of(2017, 10, 1, 10, 0, 0); -// sessionsStore.trackSessionClose(sessionCreationTime, "Sensor1"); -// -// LocalDateTime session2CreationTime = LocalDateTime.of(2017, 11, 1, 10, 0, 0); -// sessionsStore.trackSessionClose(session2CreationTime, "Sensor2"); -// -// //Exercise -// LocalDateTime queryPin = LocalDateTime.of(2017, 10, 20, 10, 0, 0); -// Set sessions = sessionsStore.sessionOlderThan(queryPin); -// -// //Verify -// assertThat(sessions).contains("Sensor1"); -// } -// -// @Test -// public void testListAllSessionsOlderThan() { -// H2SessionsStore sessionsStore = (H2SessionsStore) storageService.sessionsStore(); -// LocalDateTime sessionCreationTime = LocalDateTime.of(2017, 10, 1, 10, 0, 0); -// sessionsStore.trackSessionClose(sessionCreationTime, "Sensor1"); -// -// LocalDateTime session2CreationTime = LocalDateTime.of(2017, 10, 2, 10, 0, 0); -// sessionsStore.trackSessionClose(session2CreationTime, "Sensor2"); -// -// //Exercise -// LocalDateTime queryPin = LocalDateTime.of(2017, 10, 20, 10, 0, 0); -// Set sessions = sessionsStore.sessionOlderThan(queryPin); -// -// //Verify -// assertThat(sessions).contains("Sensor1", "Sensor2"); -// } -// -// @Test -// public void testRetrieve1000ExpiredSessionPerformance() throws InterruptedException, ExecutionException, -// TimeoutException { -// H2SessionsStore sessionsStore = (H2SessionsStore) storageService.sessionsStore(); -// -// final ExecutorService executorService = Executors.newFixedThreadPool(2); -// final Future future1 = executorService.submit(inserterForOffset(sessionsStore, 0)); -// final Future future2 = executorService.submit(inserterForOffset(sessionsStore, 50000)); -// future1.get(2, TimeUnit.SECONDS); -// future2.get(2, TimeUnit.SECONDS); -//// //prepare the fixture -//// for (int i = 0; i < 100000; i++) { -//// final LocalDateTime timePin = now().minusSeconds(1); -//// sessionsStore.trackSessionClose(timePin, "Sensor" + i); -//// } -// -// //Exercise -// LOG.warn("Start search"); -// Set sessions = sessionsStore.sessionOlderThan(now()); -// LOG.warn("Finish search"); -// -// //Verify -// assertThat(sessions.size()).isEqualTo(100000); -// } -// -// private Runnable inserterForOffset(H2SessionsStore sessionsStore, int start) { -// return () -> { -// //prepare the fixture -// for (int i = start; i < start + 50000; i++) { -// final LocalDateTime timePin = now().minusSeconds(1); -// sessionsStore.trackSessionClose(timePin, "Sensor" + i); -// } -// }; -// } -} diff --git a/settings.gradle b/settings.gradle index 4e16aab9..df367032 100644 --- a/settings.gradle +++ b/settings.gradle @@ -1,10 +1,8 @@ rootProject.name = 'moquette-parent' include ':moquette-broker' include ':moquette-distribution' -include ':moquette-h2-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