diff --git a/broker/build.gradle b/broker/build.gradle index 4a0d96ea..517facfd 100644 --- a/broker/build.gradle +++ b/broker/build.gradle @@ -33,6 +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' 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/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index ffa93727..190d6b95 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -27,9 +27,9 @@ public final class BrokerConstants { public static final String PASSWORD_FILE_PROPERTY_NAME = "password_file"; public static final String PORT_PROPERTY_NAME = "port"; public static final String HOST_PROPERTY_NAME = "host"; - public static final String DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME = "moquette_store.mapdb"; + public static final String DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME = "moquette_store.h2"; public static final String DEFAULT_PERSISTENT_PATH = System.getProperty("user.dir") + File.separator - + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; + + DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME; public static final String WEB_SOCKET_PORT_PROPERTY_NAME = "websocket_port"; public static final String WSS_PORT_PROPERTY_NAME = "secure_websocket_port"; diff --git a/broker/src/main/java/io/moquette/broker/Server.java b/broker/src/main/java/io/moquette/broker/Server.java index aba97488..4c5a6e46 100644 --- a/broker/src/main/java/io/moquette/broker/Server.java +++ b/broker/src/main/java/io/moquette/broker/Server.java @@ -52,6 +52,7 @@ public class Server { private volatile boolean initialized; private PostOffice dispatcher; private BrokerInterceptor interceptor; + private H2Builder h2Builder; public static void main(String[] args) throws IOException { final Server server = new Server(); @@ -162,7 +163,8 @@ public class Server { final ISubscriptionsRepository subscriptionsRepository; if (persistencePath != null && !persistencePath.isEmpty()) { LOG.trace("Configuring H2 subscriptions store to {}", persistencePath); - subscriptionsRepository = new H2Builder(config, scheduler).initStore().subscriptionsRepository(); + h2Builder = new H2Builder(config, scheduler); + subscriptionsRepository = h2Builder.initStore().subscriptionsRepository(); } else { LOG.trace("Configuring in-memory subscriptions store"); subscriptionsRepository = new MemorySubscriptionsRepository(); @@ -324,6 +326,11 @@ public class Server { // and SessionsRepository does not stop its tasks. Thus shutdownNow(). scheduler.shutdownNow(); + if (h2Builder != null) { + LOG.trace("Shutting down H2 persistence {}"); + h2Builder.closeStore(); + } + LOG.info("Moquette integration has been stopped."); } diff --git a/broker/src/main/java/io/moquette/interception/BrokerInterceptor.java b/broker/src/main/java/io/moquette/interception/BrokerInterceptor.java index 52d523b1..79d7cb09 100644 --- a/broker/src/main/java/io/moquette/interception/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/interception/BrokerInterceptor.java @@ -160,7 +160,7 @@ public final class BrokerInterceptor implements Interceptor { public void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg) { for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) { LOG.debug("Notifying MQTT ACK message to interceptor. CId={}, messageId={}, topic={}, interceptorId={}", - msg.getMsg().getClientID(), msg.getPacketID(), msg.getTopic(), handler.getID()); + msg.getMsg()/*.getClientID()*/, msg.getPacketID(), msg.getTopic(), handler.getID()); executor.execute(() -> handler.onMessageAcknowledged(msg)); } } diff --git a/broker/src/main/java/io/moquette/persistence/H2Builder.java b/broker/src/main/java/io/moquette/persistence/H2Builder.java index a70e5d6e..324528c1 100644 --- a/broker/src/main/java/io/moquette/persistence/H2Builder.java +++ b/broker/src/main/java/io/moquette/persistence/H2Builder.java @@ -48,4 +48,8 @@ public class H2Builder { public ISubscriptionsRepository subscriptionsRepository() { return new H2SubscriptionsRepository(mvStore); } + + public void closeStore() { + mvStore.close(); + } } diff --git a/broker/src/test/java/io/moquette/integration/IntegrationUtils.java b/broker/src/test/java/io/moquette/integration/IntegrationUtils.java index b9583dff..70d04ac6 100644 --- a/broker/src/test/java/io/moquette/integration/IntegrationUtils.java +++ b/broker/src/test/java/io/moquette/integration/IntegrationUtils.java @@ -18,54 +18,37 @@ package io.moquette.integration; import java.io.File; import java.util.Properties; -import static io.moquette.BrokerConstants.DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; +import static io.moquette.BrokerConstants.DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME; import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME; import static io.moquette.BrokerConstants.PORT_PROPERTY_NAME; +import static org.junit.Assert.assertFalse; /** * Used to carry integration configurations. */ public final class IntegrationUtils { - static String localMapDBPath() { + static String localH2MvStoreDBPath() { String currentDir = System.getProperty("user.dir"); - return currentDir + File.separator + "target" + File.separator + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; - } - - static String localClusterMapDBPath(int port) { - String currentDir = System.getProperty("user.dir"); - return currentDir + File.separator + "target" + File.separator + port + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; + return currentDir + File.separator + "target" + File.separator + DEFAULT_MOQUETTE_STORE_H2_DB_FILENAME; } public static Properties prepareTestProperties() { Properties testProperties = new Properties(); - testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath()); + testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localH2MvStoreDBPath()); testProperties.put(PORT_PROPERTY_NAME, "1883"); return testProperties; } - public static Properties prepareTestClusterProperties(int port) { - Properties testProperties = new Properties(); - testProperties.put(PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localClusterMapDBPath(port)); - testProperties.put(PORT_PROPERTY_NAME, Integer.toString(port)); - return testProperties; - } - - /*public static void cleanPersistenceFile(IConfig config) { - String fileName = config.getProperty(PERSISTENT_STORE_PROPERTY_NAME); - cleanPersistenceFile(fileName); - } - - public static void cleanPersistenceFile(String fileName) { - File dbFile = new File(fileName); - if (dbFile.exists()) { - dbFile.delete(); - new File(fileName + ".p").delete(); - new File(fileName + ".t").delete(); - } - assertFalse(dbFile.exists()); - }*/ - private IntegrationUtils() { } + + public static void clearTestStorage() { + String dbPath = localH2MvStoreDBPath(); + File dbFile = new File(dbPath); + if (dbFile.exists()) { + dbFile.delete(); + } + assertFalse(dbFile.exists()); + } } diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationDBAuthenticatorTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationDBAuthenticatorTest.java index 21bcfa81..0e49433c 100644 --- a/broker/src/test/java/io/moquette/integration/ServerIntegrationDBAuthenticatorTest.java +++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationDBAuthenticatorTest.java @@ -32,6 +32,8 @@ import java.io.IOException; import java.security.NoSuchAlgorithmException; import java.sql.SQLException; import java.util.Properties; + +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; public class ServerIntegrationDBAuthenticatorTest { @@ -99,6 +101,8 @@ public class ServerIntegrationDBAuthenticatorTest { } stopServer(); + + IntegrationUtils.clearTestStorage(); } @AfterClass diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationFuseTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationFuseTest.java index 0edbdc79..25b43608 100644 --- a/broker/src/test/java/io/moquette/integration/ServerIntegrationFuseTest.java +++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationFuseTest.java @@ -68,6 +68,8 @@ public class ServerIntegrationFuseTest { } m_server.stopServer(); + + IntegrationUtils.clearTestStorage(); } @Test diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationOpenSSLTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationOpenSSLTest.java index 421551c7..d4e9c743 100644 --- a/broker/src/test/java/io/moquette/integration/ServerIntegrationOpenSSLTest.java +++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationOpenSSLTest.java @@ -55,7 +55,7 @@ public class ServerIntegrationOpenSSLTest extends ServerIntegrationSSLTest { sslProps.put(BrokerConstants.JKS_PATH_PROPERTY_NAME, "serverkeystore.jks"); sslProps.put(BrokerConstants.KEY_STORE_PASSWORD_PROPERTY_NAME, "passw0rdsrv"); sslProps.put(BrokerConstants.KEY_MANAGER_PASSWORD_PROPERTY_NAME, "passw0rdsrv"); - sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath()); + sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localH2MvStoreDBPath()); m_server.startServer(sslProps); } } diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoTest.java index 0316efd1..c165cbf8 100644 --- a/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoTest.java +++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationPahoTest.java @@ -80,6 +80,8 @@ public class ServerIntegrationPahoTest { } stopServer(); + + IntegrationUtils.clearTestStorage(); } private void stopServer() { diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java index 714761ba..5deaf8a0 100644 --- a/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java +++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationRestartTest.java @@ -82,6 +82,8 @@ public class ServerIntegrationRestartTest { } m_server.stopServer(); + + IntegrationUtils.clearTestStorage(); } @Test diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLClientAuthTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLClientAuthTest.java index 41f84039..4b6662c1 100644 --- a/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLClientAuthTest.java +++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLClientAuthTest.java @@ -142,14 +142,14 @@ public class ServerIntegrationSSLClientAuthTest { sslProps.put(BrokerConstants.JKS_PATH_PROPERTY_NAME, "signedserverkeystore.jks"); sslProps.put(BrokerConstants.KEY_STORE_PASSWORD_PROPERTY_NAME, "passw0rdsrv"); sslProps.put(BrokerConstants.KEY_MANAGER_PASSWORD_PROPERTY_NAME, "passw0rdsrv"); - sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath()); + sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localH2MvStoreDBPath()); sslProps.put(BrokerConstants.NEED_CLIENT_AUTH, "true"); m_server.startServer(sslProps); } @Before public void setUp() throws Exception { - String dbPath = IntegrationUtils.localMapDBPath(); + String dbPath = IntegrationUtils.localH2MvStoreDBPath(); File dbFile = new File(dbPath); assertFalse(String.format("The DB storagefile %s already exists", dbPath), dbFile.exists()); @@ -171,12 +171,7 @@ public class ServerIntegrationSSLClientAuthTest { if (m_server != null) { m_server.stopServer(); } - String dbPath = IntegrationUtils.localMapDBPath(); - File dbFile = new File(dbPath); - if (dbFile.exists()) { - dbFile.delete(); - } - assertFalse(dbFile.exists()); + IntegrationUtils.clearTestStorage(); } @Test diff --git a/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLTest.java b/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLTest.java index 3d27cb72..032102d3 100644 --- a/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLTest.java +++ b/broker/src/test/java/io/moquette/integration/ServerIntegrationSSLTest.java @@ -90,13 +90,13 @@ public class ServerIntegrationSSLTest { sslProps.put(BrokerConstants.JKS_PATH_PROPERTY_NAME, "serverkeystore.jks"); sslProps.put(BrokerConstants.KEY_STORE_PASSWORD_PROPERTY_NAME, "passw0rdsrv"); sslProps.put(BrokerConstants.KEY_MANAGER_PASSWORD_PROPERTY_NAME, "passw0rdsrv"); - sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localMapDBPath()); + sslProps.put(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, IntegrationUtils.localH2MvStoreDBPath()); m_server.startServer(sslProps); } @Before public void setUp() throws Exception { - String dbPath = IntegrationUtils.localMapDBPath(); + String dbPath = IntegrationUtils.localH2MvStoreDBPath(); File dbFile = new File(dbPath); assertFalse(String.format("The DB storagefile %s already exists", dbPath), dbFile.exists()); @@ -118,12 +118,7 @@ public class ServerIntegrationSSLTest { if (m_server != null) { m_server.stopServer(); } - String dbPath = IntegrationUtils.localMapDBPath(); - File dbFile = new File(dbPath); - if (dbFile.exists()) { - dbFile.delete(); - } - assertFalse(dbFile.exists()); + IntegrationUtils.clearTestStorage(); } @Test diff --git a/distribution/build.gradle b/distribution/build.gradle index 55b7ad3b..594ba1ad 100644 --- a/distribution/build.gradle +++ b/distribution/build.gradle @@ -9,7 +9,7 @@ mainClassName = "io.moquette.integration.Server" dependencies { compile project(':moquette-broker') compile project(':moquette-h2-storage') - compile project(':moquette-mapdb-storage') + //compile project(':moquette-mapdb-storage') compile group: 'org.slf4j', name: 'slf4j-log4j12', version:'1.7.5' } diff --git a/mapdb_storage/build.gradle b/mapdb_storage/build.gradle deleted file mode 100644 index d6d109c6..00000000 --- a/mapdb_storage/build.gradle +++ /dev/null @@ -1,10 +0,0 @@ - -description = 'Moquette - MapDB storage' - - -dependencies { - compile project(':moquette-broker') - compile group: 'org.mapdb', name: 'mapdb', version:'1.0.8' - - testCompile project(path:':moquette-broker', configuration: 'archives') -} diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java deleted file mode 100644 index 28faf816..00000000 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBMessagesStore.java +++ /dev/null @@ -1,85 +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.mapdb; - -import io.moquette.spi.IMatchingCondition; -import io.moquette.spi.IMessagesStore; -import io.moquette.broker.subscriptions.Topic; -import org.mapdb.DB; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentMap; - -/** - * IMessagesStore implementation backed by MapDB. - */ -class MapDBMessagesStore implements IMessagesStore { - - private static final Logger LOG = LoggerFactory.getLogger(MapDBMessagesStore.class); - - private DB m_db; - - private ConcurrentMap m_retainedStore; - - MapDBMessagesStore(DB db) { - m_db = db; - } - - @Override - public void initStore() { - m_retainedStore = m_db.getHashMap("retained"); - LOG.info("Initialized store"); - } - - @Override - public Collection searchMatching(IMatchingCondition condition) { - LOG.debug("Scanning retained messages"); - List results = new ArrayList<>(); - for (Map.Entry entry : m_retainedStore.entrySet()) { - StoredMessage storedMsg = entry.getValue(); - if (condition.match(entry.getKey())) { - results.add(storedMsg); - } - } - - if (LOG.isTraceEnabled()) { - LOG.trace("Retained messages have been scanned matchingMessages={}", results); - } - - return results; - } - - @Override - public void cleanRetained(Topic topic) { - LOG.debug("Cleaning retained messages. Topic={}", topic); - m_retainedStore.remove(topic); - } - - @Override - public void storeRetained(Topic topic, StoredMessage storedMessage) { - LOG.debug("Store retained message for topic={}, CId={}", topic, storedMessage.getClientID()); - if (storedMessage.getClientID() == null) { - throw new IllegalArgumentException("Message to be persisted must have a not null client ID"); - } - m_retainedStore.put(topic, storedMessage); - } -} diff --git a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java b/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java deleted file mode 100644 index 8606c3ac..00000000 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBPersistentStore.java +++ /dev/null @@ -1,126 +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.mapdb; - -import io.moquette.BrokerConstants; -import io.moquette.broker.config.IConfig; -import io.moquette.spi.IMessagesStore; -import io.moquette.spi.ISessionsStore; -import io.moquette.spi.IStore; -import org.mapdb.DB; -import org.mapdb.DBMaker; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.File; -import java.io.IOException; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; - -/** - * MapDB main persistence implementation - */ -public class MapDBPersistentStore implements IStore { - - private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class); - - private DB m_db; - private final String m_storePath; - private final int m_autosaveInterval; // in seconds - - protected final ScheduledExecutorService m_scheduler; - private IMessagesStore m_messageStore; - private ISessionsStore m_sessionsStore; - - public MapDBPersistentStore(IConfig props, ScheduledExecutorService scheduler) { - this.m_storePath = props.getProperty(BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME, ""); - this.m_autosaveInterval = Integer - .parseInt(props.getProperty(BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME, "30")); - this.m_scheduler = scheduler; - } - - @Override - public IMessagesStore messagesStore() { - return m_messageStore; - } - - @Override - public ISessionsStore sessionsStore() { - return m_sessionsStore; - } - - @SuppressWarnings("FutureReturnValueIgnored") - @Override - public void initStore() { - LOG.info("Initializing MapDB store"); - if (m_storePath == null || m_storePath.isEmpty()) { - LOG.warn("MapDB store file path is empty, using in-memory store"); - m_db = DBMaker.newMemoryDB().make(); - } else { - File tmpFile; - try { - LOG.info("Using user-defined MapDB store file. Path={}", m_storePath); - tmpFile = new File(m_storePath); - boolean fileNewlyCreated = tmpFile.createNewFile(); - LOG.warn("Using {} MapDB store file. Path={}", fileNewlyCreated ? "fresh" : "existing", m_storePath); - } catch (IOException ex) { - LOG.error("Unable to open MapDB store file. Path={}, cause={}, errorMessage={}", m_storePath, - ex.getCause(), ex.getMessage()); - throw new RuntimeException("Can't create temp subscriptions file storage [" + m_storePath + "]", ex); - } - m_db = DBMaker.newFileDB(tmpFile).make(); - } - LOG.info("Scheduling MapDB commit task"); - m_scheduler.scheduleWithFixedDelay(() -> { - LOG.debug("Committing to MapDB"); - m_db.commit(); - }, this.m_autosaveInterval, this.m_autosaveInterval, TimeUnit.SECONDS); - - // TODO check m_db is valid and - m_messageStore = new MapDBMessagesStore(m_db); - m_messageStore.initStore(); - - m_sessionsStore = new MapDBSessionsStore(m_db); - m_sessionsStore.initStore(); - } - - @Override - public void close() { - if (this.m_db.isClosed()) { - LOG.warn("MapDB store is already closed. Nothing will be done"); - return; - } - LOG.info("Performing last commit to MapDB"); - this.m_db.commit(); - LOG.info("Closing MapDB store"); - this.m_db.close(); - LOG.info("Stopping MapDB commit tasks"); - - //TODO the scheduler must be stopped by the owning (the instance of Server) - //invalidate the added task - this.m_scheduler.shutdown(); - try { - m_scheduler.awaitTermination(10L, TimeUnit.SECONDS); - } catch (InterruptedException e) { - } - if (!m_scheduler.isTerminated()) { - LOG.warn("Forcing shutdown of MapDB commit tasks"); - m_scheduler.shutdown(); - } - LOG.info("MapDB store has been closed successfully"); - } -} 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 deleted file mode 100644 index 2ecfb131..00000000 --- a/mapdb_storage/src/main/java/io/moquette/persistence/mapdb/MapDBSessionsStore.java +++ /dev/null @@ -1,321 +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.mapdb; - -import io.moquette.persistence.PersistentSession; -import io.moquette.spi.IMessagesStore.StoredMessage; -import io.moquette.spi.ISessionsStore; -import io.moquette.spi.ISubscriptionsStore; -import io.moquette.broker.subscriptions.Subscription; -import io.moquette.broker.subscriptions.Topic; -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. - */ -class MapDBSessionsStore implements ISessionsStore, ISubscriptionsStore { - - private static final Logger LOG = LoggerFactory.getLogger(MapDBSessionsStore.class); - - // maps clientID->[MessageId -> msg] - private ConcurrentMap> outboundFlightMessages; - // map clientID <-> set of currently in flight packet identifiers - private Map> m_inFlightIds; - private ConcurrentMap m_persistentSessions; - // maps clientID->[MessageId -> guid] - private ConcurrentMap> m_secondPhaseStore; - - private final DB m_db; - private ConcurrentNavigableMap> sessionsClosingTimes; - - MapDBSessionsStore(DB db) { - m_db = db; - } - - @Override - public void initStore() { - outboundFlightMessages = m_db.getHashMap("outboundFlight"); - 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 - public ISubscriptionsStore subscriptionStore() { - return this; - } - - @Override - public void addNewSubscription(Subscription newSubscription) { - LOG.info("Adding new subscription. ClientId={}, topics={}", newSubscription.getClientId(), - newSubscription.getTopicFilter()); - final String clientID = newSubscription.getClientId(); - m_db.getHashMap("subscriptions_" + clientID).put(newSubscription.getTopicFilter(), newSubscription); - - if (LOG.isTraceEnabled()) { - LOG.trace("Subscription has been added. ClientId={}, topics={}, clientSubscriptions={}", - newSubscription.getClientId(), newSubscription.getTopicFilter(), - m_db.getHashMap("subscriptions_" + clientID)); - } - } - - @Override - public void removeSubscription(Topic topicFilter, String clientID) { - LOG.info("Removing subscription. ClientId={}, topics={}", clientID, topicFilter); - if (!m_db.exists("subscriptions_" + clientID)) { - return; - } - m_db.getHashMap("subscriptions_" + clientID).remove(topicFilter); - if (LOG.isDebugEnabled()) { - LOG.debug("Subscription has been removed. ClientId={}, topics={}, clientSubscriptions={}", clientID, - topicFilter, m_db.getHashMap("subscriptions_" + clientID)); - } - } - - @Override - public void wipeSubscriptions(String clientID) { - LOG.info("Wiping subscriptions. CId={}", clientID); - m_db.delete("subscriptions_" + clientID); - if (LOG.isDebugEnabled()) { - LOG.debug("Subscriptions have been removed. ClientId={}, clientSubscriptions={}", clientID, - m_db.getHashMap("subscriptions_" + clientID)); - } - } - - @Override - public List listAllSubscriptions() { - LOG.debug("Retrieving existing subscriptions..."); - List subscriptions = new ArrayList<>(); - for (String clientID : m_persistentSessions.keySet()) { - ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); - subscriptions.addAll(clientSubscriptions.values()); - } - LOG.debug("Existing subscriptions has been retrieved Result={}", subscriptions); - return subscriptions; - } - - @Override - public Collection listClientSubscriptions(String clientID) { - ConcurrentMap clientSubscriptions = m_db.getHashMap("subscriptions_" + clientID); - if (clientSubscriptions == null) { - throw new IllegalStateException("Asking for subscriptions of not persisted client: " + clientID); - } - return clientSubscriptions.values(); - } - - @Override - public Subscription reload(Subscription subscription) { - final String clientId = subscription.getClientId(); - final String clientSubcMapName = "subscriptions_" + clientId; - ConcurrentMap clientSubscriptions = m_db.getHashMap(clientSubcMapName); - LOG.debug("Retrieving subscriptions. CId={}, subscriptions={}", clientId, clientSubscriptions); - return clientSubscriptions.get(subscription.getTopicFilter()); - } - - @Override - public boolean contains(String clientID) { - return m_persistentSessions.containsKey(clientID); - } - - @Override - public void createNewDurableSession(String clientID) { - m_persistentSessions.putIfAbsent(clientID, new PersistentSession(clientID, false)); - } - - @Override - public void removeDurableSession(String clientId) { - this.m_persistentSessions.remove(clientId); - this.wipeSubscriptions(clientId); - } - - @Override - public void updateCleanStatus(String clientId, boolean newCleanStatus) { - PersistentSession updatedSession = new PersistentSession(clientId, newCleanStatus); - m_persistentSessions.put(clientId, updatedSession); - } - - @Override - public PersistentSession loadSessionByKey(String clientID) { - return this.m_persistentSessions.get(clientID); - } - - @Override - public Collection listAllSessions() { - return this.m_persistentSessions.values(); - } - - /** - * Return the next valid packetIdentifier for the given client session. - */ - @Override - public int nextPacketID(String clientID) { - LOG.debug("Generating next packet ID CId={}", clientID); - Set inFlightForClient = this.m_inFlightIds.get(clientID); - if (inFlightForClient == null) { - int nextPacketId = 1; - inFlightForClient = Collections.newSetFromMap(new ConcurrentHashMap<>()); - inFlightForClient.add(nextPacketId); - this.m_inFlightIds.put(clientID, inFlightForClient); - return nextPacketId; - } - - int maxId = inFlightForClient.isEmpty() ? 0 : Collections.max(inFlightForClient); - int nextPacketId = (maxId % 0xFFFF) + 1; - inFlightForClient.add(nextPacketId); - LOG.debug("Next packet ID has been generated CId={}, result={}", clientID, nextPacketId); - return nextPacketId; - } - - @Override - public StoredMessage inFlightAck(String clientID, int messageID) { - LOG.debug("Acknowledging inflight message CId={}, messageId={}", clientID, messageID); - ConcurrentMap m = this.outboundFlightMessages.get(clientID); - if (m == null) { - LOG.error("Can't find the inFlight record for client <{}>", clientID); - throw new RuntimeException("Can't find the inFlight record for client <" + clientID + ">"); - } - StoredMessage msg = m.remove(messageID); - this.outboundFlightMessages.put(clientID, m); - - // remove from the ids store - Set inFlightForClient = this.m_inFlightIds.get(clientID); - if (inFlightForClient != null) { - inFlightForClient.remove(messageID); - } - return msg; - } - - @Override - public void inFlight(String clientID, int messageID, StoredMessage msg) { - ConcurrentMap messages = outboundFlightMessages.get(clientID); - if (messages == null) { - messages = new ConcurrentHashMap<>(); - } - messages.put(messageID, msg); - outboundFlightMessages.put(clientID, messages); - } - - @Override - public Queue queue(String clientID) { - LOG.info("Queuing pending message. CId={}, guid={}", clientID); - return this.m_db.getQueue(clientID); - } - - @Override - public void dropQueue(String clientID) { - LOG.info("Removing pending messages. CId={}", clientID); - this.m_db.delete(clientID); - outboundFlightMessages.remove(clientID); - m_inFlightIds.remove(clientID); - } - - @Override - public void moveInFlightToSecondPhaseAckWaiting(String clientID, int messageID, StoredMessage msg) { - LOG.debug("Moving inflight message to 2nd phase ack state. CId={}, messageID={}", clientID, messageID); - ConcurrentMap m = this.m_secondPhaseStore.get(clientID); - if (m == null) { - String error = String.format("Can't find the inFlight record for client <%s> during the second phase of " + - "QoS2 pub", clientID); - LOG.error(error); - throw new RuntimeException(error); - } - m.put(messageID, msg); - this.outboundFlightMessages.put(clientID, m); - } - - @Override - public StoredMessage completeReleasedPublish(String clientID, int messageID) { - LOG.debug("Processing second phase ACK CId={}, messageId={}", clientID, messageID); - final ConcurrentMap m = this.m_secondPhaseStore.get(clientID); - if (m == null) { - String error = String.format("Can't find the inFlight record for client <%s> during the second phase " + - "acking of QoS2 pub", clientID); - LOG.error(error); - throw new RuntimeException(error); - } - - StoredMessage msg = m.remove(messageID); - m_secondPhaseStore.put(clientID, m); - return msg; - } - - @Override - public int getInflightMessagesNo(String clientID) { - int totalInflight = 0; - final String infligthMessageMapName = inboundMessageId2MessagesMapName(clientID); - ConcurrentMap inflightPerClient = m_db.getHashMap(infligthMessageMapName); - if (inflightPerClient != null) { - totalInflight += inflightPerClient.size(); - } - -// totalInflight += countPubReleaseWaitingPubComplete(clientID); - - Map outboundPerClient = outboundFlightMessages.get(clientID); - if (outboundPerClient != null) { - totalInflight += outboundPerClient.size(); - } - - return totalInflight; - } - - @Override - public int countPubReleaseWaitingPubComplete(String clientID) { - if (!m_secondPhaseStore.containsKey(clientID)) - return 0; - return m_secondPhaseStore.get(clientID).size(); - } - - @Override - public void removeTemporaryQoS2(String clientID) { - LOG.info("Removing stored messages with QoS 2. ClientId={}", clientID); - m_secondPhaseStore.remove(clientID); - } - - 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 deleted file mode 100644 index 7d5b689c..00000000 --- a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBPersistentStoreTest.java +++ /dev/null @@ -1,108 +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.mapdb; - -/** - * - * @author andrea - */ -public class MapDBPersistentStoreTest /*extends MessageStoreTCK*/ { - -// MapDBPersistentStore m_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); -// m_storageService = new MapDBPersistentStore(conf, scheduler); -// m_storageService.initStore(); -// messagesStore = m_storageService.messagesStore(); -// sessionsStore = m_storageService.sessionsStore(); -// this.sessionsRepository = new SessionsRepository(sessionsStore, null); -// } -// -// @After -// public void tearDown() { -// if (m_storageService != null) { -// m_storageService.close(); -// } -// -// scheduler.shutdown(); -// cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH); -// } -// -// static void cleanPersistenceFile(String fileName) { -// File dbFile = new File(fileName); -// if (dbFile.exists()) { -// dbFile.delete(); -// new File(fileName + ".p").delete(); -// new File(fileName + ".t").delete(); -// } -// assertFalse(dbFile.exists()); -// } -// -// @Test -// public void testNextPacketID_notExistingClientSession() { -// int packetId = sessionsStore.nextPacketID("NOT_EXISTING_CLI"); -// assertEquals(1, packetId); -// } -// -// @Test -// public void testNextPacketID_existingClientSession() { -// // Force creation of inflight map for the CLIENT session -// int packetId = sessionsStore.nextPacketID("CLIENT"); -// assertEquals(1, packetId); -// -// // request a second packetID -// packetId = sessionsStore.nextPacketID("CLIENT"); -// assertEquals(2, packetId); -// } -// -// @Test -// public void testNextPacketID() { -// StoredMessage msgStored = new StoredMessage("Hello".getBytes(UTF_8), MqttQoS.AT_LEAST_ONCE, "/topic"); -// msgStored.setClientID(TEST_CLIENT); -// -// // request a first ID -// int packetId = sessionsStore.nextPacketID("CLIENT"); -// sessionsStore.inFlight("CLIENT", packetId, msgStored); // simulate an inflight -// assertEquals(1, packetId); -// -// // release the ID -// sessionsStore.inFlightAck("CLIENT", packetId); -// -// // request a second packetID, counter restarts from 0 -// packetId = sessionsStore.nextPacketID("CLIENT"); -// assertEquals(1, packetId); -// } -// -// @Test -// public void testCloseShutdownCommitTask() throws InterruptedException { -// m_storageService.close(); -// -// // verify the executor is shutdown -// assertTrue("Storage service scheduler can't be stopped in 3 seconds", -// m_storageService.m_scheduler.awaitTermination(3, TimeUnit.SECONDS)); -// assertTrue(m_storageService.m_scheduler.isTerminated()); -// } -} 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 deleted file mode 100644 index 94ec128b..00000000 --- a/mapdb_storage/src/test/java/io/moquette/persistence/mapdb/MapDBSessionsStoreTest.java +++ /dev/null @@ -1,121 +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.mapdb; - -//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); -// } -// }; -// } - -}