diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index ba82030d..17920991 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/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; + } } 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 28453004..d79cfd68 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -98,22 +98,24 @@ public class ProtocolProcessorBootstrapper { messagesStore = store.messagesStore(); m_sessionsStore = store.sessionsStore(); storeShutdown = new Runnable() { - @Override - public void run() { - store.close(); - } - }; + + @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(); - } - }; + + @Override + public void run() { + m_mapStorage.close(); + } + }; } LOG.info("Configuring message interceptors..."); @@ -282,7 +284,7 @@ public class ProtocolProcessorBootstrapper { public ISessionsStore getSessionsStore() { return m_sessionsStore; } - + public List getSubscriptions() { return m_sessionsStore.getSubscriptions(); } 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 6a568e92..7d01ce95 100644 --- a/broker/src/main/java/io/moquette/spi/persistence/MapDBPersistentStore.java +++ b/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; } /** 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 cea1b6a3..decfd51e 100644 --- a/broker/src/test/java/io/moquette/spi/persistence/MapDBPersistentStoreTest.java +++ b/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); }