diff --git a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java index c1c7306c..05c0b9e0 100644 --- a/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java +++ b/broker/src/main/java/io/moquette/persistence/MemorySessionStore.java @@ -25,10 +25,9 @@ import io.moquette.spi.impl.subscriptions.Topic; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.time.LocalDateTime; import java.util.*; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.*; public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore { @@ -51,6 +50,7 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore { } private final Map sessions = new ConcurrentHashMap<>(); + private ConcurrentNavigableMap> sessionsClosingTimes = new ConcurrentSkipListMap<>(); MemorySessionStore() { } @@ -280,4 +280,24 @@ public class MemorySessionStore implements ISessionsStore, ISubscriptionsStore { // TODO this missing last step breaks the junit test //sessions.remove(clientID); } + + @Override + public synchronized void trackSessionClose(LocalDateTime when, String clientID) { + this.sessionsClosingTimes.putIfAbsent(when, new HashSet<>()); + this.sessionsClosingTimes.computeIfPresent(when, (key, oldSet) -> { + oldSet.add(clientID); + return oldSet; + }); + } + + @Override + public Set sessionOlderThan(LocalDateTime queryPin) { + final Set results = new HashSet<>(); + LocalDateTime keyBefore = this.sessionsClosingTimes.lowerKey(queryPin); + while (keyBefore != null) { + results.addAll(this.sessionsClosingTimes.get(keyBefore)); + keyBefore = this.sessionsClosingTimes.lowerKey(keyBefore); + } + return results; + } } diff --git a/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java b/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java index 0f1e7b0e..fabfe547 100644 --- a/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java +++ b/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java @@ -28,6 +28,8 @@ public class MemoryStorageService implements IStore { private MemorySessionStore m_sessionsStore; private MemoryMessagesStore m_messagesStore; + // NB these params must be here "by contract" used in introspection instantiation used in + // ProtocolProcessorBootstrapper.instantiateConfiguredStore public MemoryStorageService(IConfig props, ScheduledExecutorService scheduler) { m_messagesStore = new MemoryMessagesStore(); m_sessionsStore = new MemorySessionStore(); diff --git a/broker/src/main/java/io/moquette/spi/ISessionsStore.java b/broker/src/main/java/io/moquette/spi/ISessionsStore.java index d1053055..c857bdb7 100644 --- a/broker/src/main/java/io/moquette/spi/ISessionsStore.java +++ b/broker/src/main/java/io/moquette/spi/ISessionsStore.java @@ -19,8 +19,10 @@ package io.moquette.spi; import io.moquette.persistence.PersistentSession; import io.moquette.spi.IMessagesStore.StoredMessage; +import java.time.LocalDateTime; import java.util.Collection; import java.util.Queue; +import java.util.Set; /** * Store used to handle the persistence of the subscriptions tree. @@ -115,4 +117,14 @@ public interface ISessionsStore { int countPubReleaseWaitingPubComplete(String clientID); void removeTemporaryQoS2(String clientID); + + /** + * Tracks the closing time of a persistent session + * */ + void trackSessionClose(LocalDateTime when, String clientID); + + /** + * List the sessions ids closed before the pin date. + * */ + Set sessionOlderThan(LocalDateTime queryPin); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java index f91eb69a..577564c0 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -97,7 +97,8 @@ public class ProtocolProcessorBootstrapper { store.initStore(); messagesStore = store.messagesStore(); m_sessionsStore = store.sessionsStore(); - SessionsRepository sessionsRepository = new SessionsRepository(this.m_sessionsStore); + SessionsRepository sessionsRepository = new SessionsRepository(this.m_sessionsStore, server.getScheduler()); + sessionsRepository.init(); this.subscriptionsStore = m_sessionsStore.subscriptionStore(); storeShutdown = new Runnable() { diff --git a/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java b/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java index 6ec89414..04c49c43 100644 --- a/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java +++ b/broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java @@ -13,7 +13,6 @@ * * You may elect to redistribute this code under either of these licenses. */ - package io.moquette.spi.impl; import io.moquette.persistence.PersistentSession; @@ -21,21 +20,44 @@ import io.moquette.spi.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.time.LocalDateTime; +import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.Collection; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; public class SessionsRepository { + /** + * Task to be scheduled to execute the cleaning of persisted sessions (clean flag=false) older than a defined period. + */ + private class SessionCleanerTask implements Runnable { + + @Override + public void run() { + wipeExpiredSessions(); + } + } + private static final Logger LOG = LoggerFactory.getLogger(SessionsRepository.class); private final ISessionsStore sessions; private ISubscriptionsStore subscriptionsStore; + private ScheduledExecutorService scheduler; private final Map sessionsCache = new ConcurrentHashMap<>(); - public SessionsRepository(ISessionsStore sessionsStore) { + public SessionsRepository(ISessionsStore sessionsStore, ScheduledExecutorService scheduler) { this.sessions = sessionsStore; this.subscriptionsStore = sessionsStore.subscriptionStore(); + this.scheduler = scheduler; + } + + public void init() { + SessionCleanerTask cleanerTask = new SessionCleanerTask(); + this.scheduler.schedule(cleanerTask, 1, TimeUnit.HOURS); } public ClientSession sessionForClient(String clientID) { @@ -122,5 +144,15 @@ public class SessionsRepository { } sessionsCache.remove(clientId); + this.sessions.trackSessionClose(LocalDateTime.now(), clientId); + } + + private void wipeExpiredSessions() { + final LocalDateTime pin = LocalDateTime.now().minus(6, ChronoUnit.DAYS); + final Set expiredSessionsIds = this.sessions.sessionOlderThan(pin); + for (String expiredSession : expiredSessionsIds) { + this.sessions.removeDurableSession(expiredSession); + this.subscriptionsStore.wipeSubscriptions(expiredSession); + } } } diff --git a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java index f22d5451..005fb795 100644 --- a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java +++ b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java @@ -41,7 +41,7 @@ public class ClientSessionTest { subscriptionsDirectory = new CTrieSubscriptionDirectory(); MemoryStorageService storageService = new MemoryStorageService(null, null); ISessionsStore sessionsStore = storageService.sessionsStore(); - sessionsRepository = new SessionsRepository(sessionsStore); + sessionsRepository = new SessionsRepository(sessionsStore, null); subscriptionsDirectory.init(sessionsRepository); } diff --git a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java index 6d7096cd..a5d83f1a 100644 --- a/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java +++ b/broker/src/test/java/io/moquette/spi/impl/AbstractProtocolProcessorCommonUtils.java @@ -68,7 +68,7 @@ abstract class AbstractProtocolProcessorCommonUtils { m_sessionStore = memStorage.sessionsStore(); // m_messagesStore.initStore(); - sessionsRepository = new SessionsRepository(m_sessionStore); + sessionsRepository = new SessionsRepository(m_sessionStore, null); Set clientIds = new HashSet<>(); clientIds.add(FAKE_CLIENT_ID); 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 4741d98b..8d530d5a 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java @@ -86,7 +86,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils // simulate a connect that register a clientID to an IoSession MemoryStorageService storageService = new MemoryStorageService(null, null); - SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore()); + SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore(), null); subs.init(sessionsRepository); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); @@ -123,7 +123,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils // simulate a connect that register a clientID to an IoSession MemoryStorageService storageService = new MemoryStorageService(null, null); - SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore()); + SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore(), null); subs.init(sessionsRepository); m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, this.sessionsRepository); @@ -246,7 +246,7 @@ public class ProtocolProcessorTest extends AbstractProtocolProcessorCommonUtils } }; MemoryStorageService storageService = new MemoryStorageService(null, null); - SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore()); + SessionsRepository sessionsRepository = new SessionsRepository(storageService.sessionsStore(), null); subs.init(sessionsRepository); // simulate a connect that register a clientID to an IoSession 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 5a640fad..949ce1de 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 @@ -65,11 +65,11 @@ public class ProtocolProcessor_CONNECT_Test { m_mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD)); subscriptions = new CTrieSubscriptionDirectory(); - SessionsRepository sessionsRepository = new SessionsRepository(m_sessionStore); + SessionsRepository sessionsRepository = new SessionsRepository(m_sessionStore, null); subscriptions.init(sessionsRepository); m_processor = new ProtocolProcessor(); m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore)); + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null)); } @Test @@ -159,7 +159,7 @@ public class ProtocolProcessor_CONNECT_Test { protected void reinitProcessorProhibitingAnonymousClients() { m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, false, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore)); + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null)); } @Test @@ -324,7 +324,7 @@ public class ProtocolProcessor_CONNECT_Test { protected void reinitProtocolProcessorWithZeroLengthClientIdAndAnonymousClients() { m_processor.init(subscriptions, m_messagesStore, m_sessionStore, m_mockAuthenticator, true, true, - new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore)); + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR, new SessionsRepository(this.m_sessionStore, null)); } @Test diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java index 78f2a486..84b53fb9 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryMatchingTest.java @@ -12,7 +12,8 @@ * http://www.opensource.org/licenses/apache2.0.php * * You may elect to redistribute this code under either of these licenses. - */package io.moquette.spi.impl.subscriptions; + */ +package io.moquette.spi.impl.subscriptions; import io.moquette.persistence.MemoryStorageService; import io.moquette.spi.ClientSession; @@ -45,7 +46,7 @@ public class CTrieSubscriptionDirectoryMatchingTest { MemoryStorageService storageService = new MemoryStorageService(null, null); ISessionsStore sessionsStore = storageService.sessionsStore(); this.subscriptionsStore = sessionsStore.subscriptionStore(); - this.sessionsRepository = new SessionsRepository(sessionsStore); + this.sessionsRepository = new SessionsRepository(sessionsStore, null); sut.init(this.sessionsRepository); } @@ -178,7 +179,7 @@ public class CTrieSubscriptionDirectoryMatchingTest { sut = new CTrieSubscriptionDirectory(); MemoryStorageService memStore = new MemoryStorageService(null, null); ISessionsStore aSessionsStore = memStore.sessionsStore(); - SessionsRepository sessionsRepository = new SessionsRepository(aSessionsStore); + SessionsRepository sessionsRepository = new SessionsRepository(aSessionsStore, null); sut.init(sessionsRepository); Subscription sub = clientSubOnTopic("AnySensor1", s); @@ -191,7 +192,7 @@ public class CTrieSubscriptionDirectoryMatchingTest { sut = new CTrieSubscriptionDirectory(); MemoryStorageService memStore = new MemoryStorageService(null, null); ISessionsStore aSessionsStore = memStore.sessionsStore(); - SessionsRepository sessionsRepository = new SessionsRepository(aSessionsStore); + SessionsRepository sessionsRepository = new SessionsRepository(aSessionsStore, null); sut.init(sessionsRepository); Subscription sub = clientSubOnTopic("AnySensor1", subscription); diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java index 954ab7a8..9fb34588 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/subscriptions/CTrieSubscriptionDirectoryTest.java @@ -37,7 +37,7 @@ public class CTrieSubscriptionDirectoryTest { sut = new CTrieSubscriptionDirectory(); MemoryStorageService memStore = new MemoryStorageService(null, null); ISessionsStore aSessionsStore = memStore.sessionsStore(); - SessionsRepository sessionsRepository = new SessionsRepository(aSessionsStore); + SessionsRepository sessionsRepository = new SessionsRepository(aSessionsStore, null); sut.init(sessionsRepository); } diff --git a/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java b/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java index 9733773c..9134eecb 100644 --- a/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java +++ b/h2_storage/src/main/java/io/moquette/persistence/h2/H2SessionsStore.java @@ -27,6 +27,7 @@ import org.h2.mvstore.MVStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.time.LocalDateTime; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -43,6 +44,7 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { private Map> inFlightIds; // maps clientID->[MessageId -> guid] private ConcurrentMap> secondPhaseStore; + private MVMap> sessionsClosingTimes; H2SessionsStore(MVStore mvStore) { this.mvStore = mvStore; @@ -54,6 +56,7 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { this.outboundFlightMessages = mvStore.openMap("outboundFlight"); this.inFlightIds = mvStore.openMap("inflightPacketIDs"); this.secondPhaseStore = mvStore.openMap("secondPhase"); + this.sessionsClosingTimes = mvStore.openMap("sessionsCreationTimes"); LOG.info("Initialized sessions H2 store"); } @@ -290,4 +293,23 @@ public class H2SessionsStore implements ISessionsStore, ISubscriptionsStore { this.secondPhaseStore.remove(clientID); } + @Override + public synchronized void trackSessionClose(LocalDateTime when, String clientID) { + this.sessionsClosingTimes.putIfAbsent(when, new HashSet<>()); + this.sessionsClosingTimes.computeIfPresent(when, (key, oldSet) -> { + oldSet.add(clientID); + return oldSet; + }); + } + + @Override + public Set sessionOlderThan(LocalDateTime queryPin) { + final Set results = new HashSet<>(); + LocalDateTime keyBefore = this.sessionsClosingTimes.lowerKey(queryPin); + while (keyBefore != null) { + results.addAll(this.sessionsClosingTimes.get(keyBefore)); + keyBefore = this.sessionsClosingTimes.lowerKey(keyBefore); + } + return results; + } } 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 index 9131d5b0..ad8fbf4a 100644 --- a/h2_storage/src/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java +++ b/h2_storage/src/test/java/io/moquette/persistence/h2/H2MessagesStoreTest.java @@ -48,7 +48,7 @@ public class H2MessagesStoreTest extends MessageStoreTCK { storageService.initStore(); messagesStore = storageService.messagesStore(); sessionsStore = storageService.sessionsStore(); - this.sessionsRepository = new SessionsRepository(sessionsStore); + this.sessionsRepository = new SessionsRepository(sessionsStore, null); } @After 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 new file mode 100644 index 00000000..6970dc25 --- /dev/null +++ b/h2_storage/src/test/java/io/moquette/persistence/h2/H2SessionsStoreTest.java @@ -0,0 +1,132 @@ +/* + * Copyright (c) 2012-2017 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 io.moquette.BrokerConstants; +import io.moquette.server.config.IConfig; +import io.moquette.server.config.MemoryConfig; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.LocalDateTime; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.*; + +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/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java index 521209cf..71ffe2f1 100644 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java +++ b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java @@ -26,9 +26,11 @@ import org.mapdb.DB; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.time.LocalDateTime; import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ConcurrentNavigableMap; /** * ISessionsStore implementation backed by MapDB. @@ -46,6 +48,7 @@ class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore { private ConcurrentMap> m_secondPhaseStore; private final DB m_db; + private ConcurrentNavigableMap> sessionsClosingTimes; MapDBSessionsStore(DB db) { m_db = db; @@ -57,6 +60,7 @@ class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore { m_inFlightIds = m_db.getHashMap("inflightPacketIDs"); m_persistentSessions = m_db.getHashMap("sessions"); m_secondPhaseStore = m_db.getHashMap("secondPhase"); + this.sessionsClosingTimes = m_db.getTreeMap("sessionsCreationTimes"); } @Override @@ -291,4 +295,25 @@ class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore { private static String inboundMessageId2MessagesMapName(String clientID) { return "inboundInflight_" + clientID; } + + @Override + public synchronized void trackSessionClose(LocalDateTime when, String clientID) { + this.sessionsClosingTimes.putIfAbsent(when, new HashSet<>()); + this.sessionsClosingTimes.computeIfPresent(when, (key, oldSet) -> { + oldSet.add(clientID); + return oldSet; + }); + } + + @Override + public Set sessionOlderThan(LocalDateTime queryPin) { + final Set results = new HashSet<>(); + LocalDateTime keyBefore = this.sessionsClosingTimes.lowerKey(queryPin); + while (keyBefore != null) { + results.addAll(this.sessionsClosingTimes.get(keyBefore)); + keyBefore = this.sessionsClosingTimes.lowerKey(keyBefore); + } + return results; + } + } diff --git a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java index 97fea12f..0643813a 100644 --- a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java +++ b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java @@ -57,7 +57,7 @@ public class MapDBPersistentStoreTest extends MessageStoreTCK { m_storageService.initStore(); messagesStore = m_storageService.messagesStore(); sessionsStore = m_storageService.sessionsStore(); - this.sessionsRepository = new SessionsRepository(sessionsStore); + this.sessionsRepository = new SessionsRepository(sessionsStore, null); } @After @@ -70,7 +70,7 @@ public class MapDBPersistentStoreTest extends MessageStoreTCK { cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); } - public static void cleanPersistenceFile(String fileName) { + static void cleanPersistenceFile(String fileName) { File dbFile = new File(fileName); if (dbFile.exists()) { dbFile.delete(); diff --git a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBSessionsStoreTest.java b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBSessionsStoreTest.java new file mode 100644 index 00000000..b52d0b4f --- /dev/null +++ b/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBSessionsStoreTest.java @@ -0,0 +1,134 @@ +/* + * Copyright (c) 2012-2017 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.mapdb; + +import io.moquette.BrokerConstants; +import io.moquette.server.config.IConfig; +import io.moquette.server.config.MemoryConfig; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.time.LocalDateTime; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.*; + +import static io.moquette.persistence.mapdb.MapDBPersistentStoreTest.cleanPersistenceFile; +import static java.time.LocalDateTime.now; +import static org.assertj.core.api.Assertions.assertThat; + +public class MapDBSessionsStoreTest { + + private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStoreTest.class); + + MapDBPersistentStore 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 MapDBPersistentStore(conf, scheduler); + storageService.initStore(); + } + + @After + public void tearDown() { + if (storageService != null) { + storageService.close(); + } + + scheduler.shutdown(); + cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); + } + + @Test + public void testQueryByExpirationDate() { + MapDBSessionsStore sessionsStore = (MapDBSessionsStore) 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() { + MapDBSessionsStore sessionsStore = (MapDBSessionsStore) 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 { + MapDBSessionsStore sessionsStore = (MapDBSessionsStore) storageService.sessionsStore(); + + final ExecutorService executorService = Executors.newFixedThreadPool(2); + final Future future1 = executorService.submit(inserterForOffset(sessionsStore, 0)); + final Future future2 = executorService.submit(inserterForOffset(sessionsStore, 5000)); + future1.get(10, TimeUnit.SECONDS); + future2.get(10, 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(10000); + } + + private Runnable inserterForOffset(MapDBSessionsStore sessionsStore, int start) { + return () -> { + //prepare the fixture + for (int i = start; i < start + 5000; i++) { + final LocalDateTime timePin = now().minusSeconds(1); + sessionsStore.trackSessionClose(timePin, "Sensor" + i); + } + }; + } + +} diff --git a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java index bedb259e..f3dc08ae 100644 --- a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java +++ b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java @@ -38,7 +38,7 @@ public final class PerformanceTest { ISubscriptionsDirectory store = new CTrieSubscriptionDirectory(); MemoryStorageService memStore = new MemoryStorageService(null, null); ISessionsStore aSessionsStore = memStore.sessionsStore(); - SessionsRepository sessionsRepository = new SessionsRepository(aSessionsStore); + SessionsRepository sessionsRepository = new SessionsRepository(aSessionsStore, null); store.init(sessionsRepository); int times = 10000;