From e3b63916531020bbe5f7ec4b64cbd3a9396bb655 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 22 Nov 2015 16:56:40 +0100 Subject: [PATCH] Added 'autosave_interval' fix issue #97 --- ChangeLog.txt | 2 ++ .../java/io/moquette/spi/impl/SimpleMessaging.java | 3 +-- .../spi/persistence/MapDBPersistentStore.java | 14 ++++++++++---- .../spi/persistence/MapDBPersistentStoreTest.java | 10 +++++++++- broker/src/test/resources/log4j.properties | 2 +- distribution/src/main/resources/moquette.conf | 11 ++++++++++- .../main/java/io/moquette/commons/Constants.java | 1 + 7 files changed, 34 insertions(+), 9 deletions(-) diff --git a/ChangeLog.txt b/ChangeLog.txt index 0e6b1d66..e3f4d13f 100644 --- a/ChangeLog.txt +++ b/ChangeLog.txt @@ -12,6 +12,8 @@ Version 0.8: Added check to avoid creation of LostConnectionEvents for not yet completed connects (fix issue #88) Removed LMAX ringbuffer to use multi threading with persistent (copy on write) data structures Fixed bug that on connection stolen dropped also the new connection (fix issue #108) + Refactored the storage part to be more rational and use ClientSession concept + Added config property 'autosave_interval' (issue #97) Version 0.7: Fix issue #38, added aligned to MQTT 3.1.1 diff --git a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java index 0bc534e2..2a091057 100644 --- a/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/io/moquette/spi/impl/SimpleMessaging.java @@ -69,8 +69,7 @@ public class SimpleMessaging { public ProtocolProcessor init(IConfig props) { subscriptions = new SubscriptionsStore(); - //TODO use a property to select the storage path - m_mapStorage = new MapDBPersistentStore(props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, "")); + m_mapStorage = new MapDBPersistentStore(props); m_mapStorage.initStore(); IMessagesStore messagesStore = m_mapStorage.messagesStore(); ISessionsStore sessionsStore = m_mapStorage.sessionsStore(messagesStore); diff --git a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java index a09a6e5b..590043ad 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java +++ b/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java @@ -16,6 +16,7 @@ package io.moquette.spi.persistence; +import io.moquette.server.config.IConfig; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.proto.MQTTException; @@ -31,6 +32,9 @@ import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import static io.moquette.commons.Constants.PERSISTENT_STORE_PROPERTY_NAME; +import static io.moquette.commons.Constants.AUTOSAVE_INTERVAL_PROPERTY_NAME; + /** * MapDB main persistence implementation */ @@ -53,12 +57,14 @@ public class MapDBPersistentStore { private static final Logger LOG = LoggerFactory.getLogger(MapDBPersistentStore.class); private DB m_db; - private String m_storePath; + private final String m_storePath; + private final int m_autosaveInterval; // in seconds protected final ScheduledExecutorService m_scheduler = Executors.newScheduledThreadPool(1); - public MapDBPersistentStore(String storePath) { - this.m_storePath = storePath; + public MapDBPersistentStore(IConfig props) { + this.m_storePath = props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, ""); + this.m_autosaveInterval = Integer.parseInt(props.getProperty(AUTOSAVE_INTERVAL_PROPERTY_NAME, "30")); } /** @@ -97,7 +103,7 @@ public class MapDBPersistentStore { public void run() { m_db.commit(); } - }, 30, 30, TimeUnit.SECONDS); + }, this.m_autosaveInterval, this.m_autosaveInterval, TimeUnit.SECONDS); } public void close() { diff --git a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java index c3cb1bdc..1cbd0b6d 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java @@ -16,7 +16,10 @@ package io.moquette.spi.persistence; import io.moquette.commons.Constants; +import static io.moquette.commons.Constants.*; import io.moquette.server.IntegrationUtils; +import io.moquette.server.config.IConfig; +import io.moquette.server.config.MemoryConfig; import io.moquette.spi.IMessagesStore; import io.moquette.spi.ISessionsStore; import io.moquette.proto.messages.AbstractMessage; @@ -25,7 +28,9 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; +import java.util.Collections; import java.util.List; +import java.util.Properties; import java.util.concurrent.TimeUnit; import static org.junit.Assert.*; @@ -43,7 +48,10 @@ public class MapDBPersistentStoreTest { @Before public void setUp() throws Exception { IntegrationUtils.cleanPersistenceFile(Constants.DEFAULT_PERSISTENT_PATH); - m_storageService = new MapDBPersistentStore(Constants.DEFAULT_PERSISTENT_PATH); + Properties props = new Properties(); + props.setProperty(PERSISTENT_STORE_PROPERTY_NAME, DEFAULT_PERSISTENT_PATH); + IConfig conf = new MemoryConfig(props); + m_storageService = new MapDBPersistentStore(conf); m_storageService.initStore(); m_messagesStore = m_storageService.messagesStore(); m_sessionsStore = m_storageService.sessionsStore(m_messagesStore); diff --git a/broker/src/test/resources/log4j.properties b/broker/src/test/resources/log4j.properties index 4d59128e..4c9235bd 100644 --- a/broker/src/test/resources/log4j.properties +++ b/broker/src/test/resources/log4j.properties @@ -1,7 +1,7 @@ # Set root logger level to DEBUG and its only appender to A1. log4j.rootLogger=ERROR, stdout, file -#og4j.logger.io.moquette=WARN +#log4j.logger.io.moquette=WARN log4j.logger.io.moquette=DEBUG log4j.logger.io.moquette.spi.impl.ProtocolProcessor=DEBUG diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf index e1a960eb..7e2ed9ea 100644 --- a/distribution/src/main/resources/moquette.conf +++ b/distribution/src/main/resources/moquette.conf @@ -75,4 +75,13 @@ password_file config/password_file.conf # If not specified uses FileAuthenticator #********************************************************************* # authenticator_class [[path to your class>]] -# authorizator_class [[path to your class>]] \ No newline at end of file +# authorizator_class [[path to your class>]] + + +#********************************************************************* +# Persistence configuration +# autosave_interval: +# interval between flushes of MapDB storage to disk. It's in +# seconds, if not specified defaults is 30 s. +#********************************************************************* +# autosave_interval 120 \ No newline at end of file diff --git a/parser_commons/src/main/java/io/moquette/commons/Constants.java b/parser_commons/src/main/java/io/moquette/commons/Constants.java index 6b4c28fa..4389d2a4 100644 --- a/parser_commons/src/main/java/io/moquette/commons/Constants.java +++ b/parser_commons/src/main/java/io/moquette/commons/Constants.java @@ -28,6 +28,7 @@ public class Constants { public static final String DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME = "moquette_store.mapdb"; public static final String DEFAULT_PERSISTENT_PATH = System.getProperty("user.dir") + File.separator + DEFAULT_MOQUETTE_STORE_MAP_DB_FILENAME; public static final String PERSISTENT_STORE_PROPERTY_NAME = "persistent_store"; + public static final String AUTOSAVE_INTERVAL_PROPERTY_NAME = "autosave_interval"; 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";