Added 'autosave_interval' fix issue #97

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

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);

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() {

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);

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

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>]]
+# 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

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";