broker/Server: ScheduledExecutorService moved from MapDBPersistentStore

broker/ProtocolProcessorBootstrapper: tabs removed
broker/MapDBPersistentStore: bonding to Server removed

broker/src/main/java/io/moquette/server/Server.java

@@ -45,6 +45,9 @@ import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+
 import static io.moquette.logging.LoggingUtils.getInterceptorIds;
 
 /**
@@ -68,6 +71,8 @@ public class Server {
 
     private ProtocolProcessorBootstrapper m_processorBootstrapper;
 
+    private ScheduledExecutorService scheduler;
+
     public static void main(String[] args) throws IOException {
         final Server server = new Server();
         server.startServer();
@@ -172,6 +177,8 @@ public class Server {
         }
         LOG.info("Starting Moquette Server. MQTT message interceptors = {}.", getInterceptorIds(handlers));
 
+        scheduler = Executors.newScheduledThreadPool(1);
+
         final String handlerProp = System.getProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME);
         if (handlerProp != null) {
             config.setProperty(BrokerConstants.INTERCEPT_HANDLER_PROPERTY_NAME, handlerProp);
@@ -270,6 +277,9 @@ public class Server {
                 LOG.warn("The embedded Hazelcast instance is already shut down.");
             }
         }
+
+        scheduler.shutdown();
+
         LOG.info("The Moquette server has been stopped.");
     }
 
@@ -332,4 +342,8 @@ public class Server {
     public ProtocolProcessor getProcessor() {
         return m_processor;
     }
+    
+    public ScheduledExecutorService getScheduler() {
+        return scheduler;
+    }
 }

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java

@@ -98,17 +98,19 @@ public class ProtocolProcessorBootstrapper {
             messagesStore = store.messagesStore();
             m_sessionsStore = store.sessionsStore();
             storeShutdown = new Runnable() {
+
                 @Override
                 public void run() {
                     store.close();
                 }
             };
         } else {
-            final MapDBPersistentStore m_mapStorage = new MapDBPersistentStore(props);
+            final MapDBPersistentStore m_mapStorage = new MapDBPersistentStore(props, server.getScheduler());
             m_mapStorage.initStore();
             messagesStore = m_mapStorage.messagesStore();
             m_sessionsStore = m_mapStorage.sessionsStore();
             storeShutdown = new Runnable() {
+
                 @Override
                 public void run() {
                     m_mapStorage.close();

broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java

@@ -27,7 +27,6 @@ import org.slf4j.LoggerFactory;
 import java.io.File;
 import java.io.IOException;
 import java.io.Serializable;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import static io.moquette.BrokerConstants.AUTOSAVE_INTERVAL_PROPERTY_NAME;
@@ -58,13 +57,14 @@ public class MapDBPersistentStore implements IStore {
     private final String m_storePath;
     private final int m_autosaveInterval; // in seconds
 
-    protected final ScheduledExecutorService m_scheduler = Executors.newScheduledThreadPool(1);
+    protected final ScheduledExecutorService m_scheduler;
     private IMessagesStore m_messageStore;
     private ISessionsStore m_sessionsStore;
 
-    public MapDBPersistentStore(IConfig props) {
+    public MapDBPersistentStore(IConfig props, ScheduledExecutorService scheduler) {
         this.m_storePath = props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, "");
         this.m_autosaveInterval = Integer.parseInt(props.getProperty(AUTOSAVE_INTERVAL_PROPERTY_NAME, "30"));
+        this.m_scheduler = scheduler;
     }
 
     /**

broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java

@@ -16,6 +16,8 @@
 
 package io.moquette.spi.persistence;
 
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import io.moquette.BrokerConstants;
 import io.moquette.server.IntegrationUtils;
 import io.moquette.server.config.IConfig;
@@ -47,13 +49,17 @@ public class MapDBPersistentStoreTest {
     ISessionsStore m_sessionsStore;
     IMessagesStore m_messagesStore;
 
+    private ScheduledExecutorService scheduler;
+
     @Before
     public void setUp() throws Exception {
+        scheduler = Executors.newScheduledThreadPool(1);
+
         IntegrationUtils.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);
+        m_storageService = new MapDBPersistentStore(conf, scheduler);
         m_storageService.initStore();
         m_messagesStore = m_storageService.messagesStore();
         m_sessionsStore = m_storageService.sessionsStore();
@@ -65,6 +71,7 @@ public class MapDBPersistentStoreTest {
             m_storageService.close();
         }
 
+        scheduler.shutdown();
         IntegrationUtils.cleanPersistenceFile(BrokerConstants.DEFAULT_PERSISTENT_PATH);
     }