broker/MapDBPersistentStore: Interfaces used for the Constructors

broker/IStore: Interface added for MapDBPersistentStore
broker/ProtocolProcessorBootstrapper: class loader for store added
broker/ISessionsStore#mapToGuid: moved to IMessagesStore
broker/ProtocolProcessorBootstrapper: getter used for SessionStore
broker/Server: getter for processor added
broker/ProtocolProcessor: getter added for Stores

broker/src/main/java/io/moquette/BrokerConstants.java

@@ -59,4 +59,5 @@ public class BrokerConstants {
     public static final String NETTY_TCP_NODELAY_PROPERTY_NAME = "netty.tcp_nodelay";
     public static final String NETTY_SO_KEEPALIVE_PROPERTY_NAME = "netty.so_keepalive";
     public static final String NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME = "netty.channel_timeout.seconds";
+    public static final String STORAGE_CLASS_NAME = "storage_class";
 }

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

@@ -329,4 +329,7 @@ public class Server {
         return m_processorBootstrapper.getConnectionDescriptors();
     }
 
+    public ProtocolProcessor getProcessor() {
+        return m_processor;
+    }
 }

broker/src/main/java/io/moquette/spi/ClientSession.java

@@ -188,7 +188,7 @@ public class ClientSession {
     }
 
     public IMessagesStore.StoredMessage storedMessage(int messageID) {
-        final MessageGUID guid = m_sessionsStore.mapToGuid(clientID, messageID);
+        final MessageGUID guid = messagesStore.mapToGuid(clientID, messageID);
         return messagesStore.getMessageByGuid(guid);
     }
 

broker/src/main/java/io/moquette/spi/IMessagesStore.java

@@ -142,4 +142,6 @@ public interface IMessagesStore {
     void cleanRetained(Topic topic);
 
     int getPendingPublishMessages(String clientID);
+
+    MessageGUID mapToGuid(String clientID, int messageID);
 }

broker/src/main/java/io/moquette/spi/ISessionsStore.java

@@ -183,8 +183,6 @@ public interface ISessionsStore {
      */
     MessageGUID secondPhaseAcknowledged(String clientID, int messageID);
 
-    MessageGUID mapToGuid(String clientID, int messageID);
-
     StoredMessage getInflightMessage(String clientID, int messageID);
 
     /**

broker/src/main/java/io/moquette/spi/IStore.java

@@ -0,0 +1,12 @@
+package io.moquette.spi;
+
+
+public interface IStore {
+
+    IMessagesStore messagesStore();
+
+    ISessionsStore sessionsStore();
+
+    void close();
+
+}

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

@@ -136,8 +136,11 @@ public class ProtocolProcessor {
     private boolean allowAnonymous;
     private boolean allowZeroByteClientId;
     private IAuthorizator m_authorizator;
+
     private IMessagesStore m_messagesStore;
+
     private ISessionsStore m_sessionsStore;
+
     private IAuthenticator m_authenticator;
     private BrokerInterceptor m_interceptor;
     private String m_server_port;
@@ -1063,4 +1066,12 @@ public class ProtocolProcessor {
     public void removeInterceptHandler(InterceptHandler interceptHandler) {
         this.m_interceptor.removeInterceptHandler(interceptHandler);
     }
+    
+    public IMessagesStore getMessagesStore() {
+        return m_messagesStore;
+    }
+    
+    public ISessionsStore getSessionsStore() {
+        return m_sessionsStore;
+    }
 }

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

@@ -24,6 +24,7 @@ import io.moquette.server.config.IConfig;
 import io.moquette.server.config.IResourceLoader;
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.ISessionsStore;
+import io.moquette.spi.IStore;
 import io.moquette.spi.impl.security.ACLFileParser;
 import io.moquette.spi.impl.security.AcceptAllAuthenticator;
 import io.moquette.spi.impl.security.DenyAllAuthorizator;
@@ -53,10 +54,10 @@ public class ProtocolProcessorBootstrapper {
 
     private SubscriptionsStore subscriptions;
 
-    private MapDBPersistentStore m_mapStorage;
-
     private ISessionsStore m_sessionsStore;
 
+    private Runnable storeShutdown = null;
+
     private BrokerInterceptor m_interceptor;
 
     private final ProtocolProcessor m_processor = new ProtocolProcessor();
@@ -89,11 +90,31 @@ public class ProtocolProcessorBootstrapper {
             IAuthenticator authenticator, IAuthorizator authorizator, Server server) {
         subscriptions = new SubscriptionsStore();
 
+        IMessagesStore messagesStore;
         LOG.info("Initializing messages and sessions stores...");
-        m_mapStorage = new MapDBPersistentStore(props);
+        String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, "");
-        m_mapStorage.initStore();
+        if (storageClassName != null && !storageClassName.isEmpty()) {
-        IMessagesStore messagesStore = m_mapStorage.messagesStore();
+            final IStore store = loadClass(storageClassName, IStore.class, Server.class, server);
-        m_sessionsStore = m_mapStorage.sessionsStore();
+            messagesStore = store.messagesStore();
+            m_sessionsStore = store.sessionsStore();
+            storeShutdown = new Runnable() {
+				@Override
+				public void run() {
+					store.close();
+				}
+			};
+        } else {
+            final MapDBPersistentStore m_mapStorage = new MapDBPersistentStore(props);
+            m_mapStorage.initStore();
+            messagesStore = m_mapStorage.messagesStore();
+            m_sessionsStore = m_mapStorage.sessionsStore();
+            storeShutdown = new Runnable() {
+				@Override
+				public void run() {
+					m_mapStorage.close();
+				}
+			};
+        }
 
         LOG.info("Configuring message interceptors...");
 
@@ -258,12 +279,17 @@ public class ProtocolProcessorBootstrapper {
         return instance;
     }
 
+    public ISessionsStore getSessionsStore() {
+        return m_sessionsStore;
+    }
+    
     public List<Subscription> getSubscriptions() {
         return m_sessionsStore.getSubscriptions();
     }
 
     public void shutdown() {
-        this.m_mapStorage.close();
+        if (storeShutdown != null)
+            storeShutdown.run();
     }
 
     public ConnectionDescriptorStore getConnectionDescriptors() {

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

@@ -91,7 +91,7 @@ class MapDBMessagesStore implements IMessagesStore {
                 storedMessage.getTopic());
         m_persistentMessageStore.put(guid, storedMessage);
         ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db
-                .getHashMap(MapDBSessionsStore.messageId2GuidsMapName(storedMessage.getClientID()));
+                .getHashMap(messageId2GuidsMapName(storedMessage.getClientID()));
         messageIdToGuid.put(storedMessage.getMessageID(), guid);
         return guid;
     }
@@ -99,8 +99,7 @@ class MapDBMessagesStore implements IMessagesStore {
     @Override
     public void dropMessagesInSession(String clientID) {
         LOG.debug("Dropping stored messages. ClientId = {}.", clientID);
-        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db
+        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
-                .getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID));
         for (MessageGUID guid : messageIdToGuid.values()) {
             removeStoredMessage(guid);
         }
@@ -136,7 +135,20 @@ class MapDBMessagesStore implements IMessagesStore {
     @Override
     public int getPendingPublishMessages(String clientID) {
         ConcurrentMap<Integer, MessageGUID> messageIdToGuidMap = m_db
-                .getHashMap(MapDBSessionsStore.messageId2GuidsMapName(clientID));
+                .getHashMap(messageId2GuidsMapName(clientID));
         return messageIdToGuidMap.size();
     }
+
+    @Override
+    public MessageGUID mapToGuid(String clientID, int messageID) {
+        LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID);
+        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
+        MessageGUID result = messageIdToGuid.get(messageID);
+        LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result);
+        return result;
+    }
+
+    static String messageId2GuidsMapName(String clientID) {
+        return "guidsMapping_" + clientID;
+    }
 }

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

@@ -19,6 +19,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.spi.IStore;
 import org.mapdb.DB;
 import org.mapdb.DBMaker;
 import org.slf4j.Logger;
@@ -35,7 +36,7 @@ import static io.moquette.BrokerConstants.PERSISTENT_STORE_PROPERTY_NAME;
 /**
  * MapDB main persistence implementation
  */
-public class MapDBPersistentStore {
+public class MapDBPersistentStore implements IStore {
 
     /**
      * This is a DTO used to persist minimal status (clean session and activation status) of a
@@ -58,8 +59,8 @@ public class MapDBPersistentStore {
     private final int m_autosaveInterval; // in seconds
 
     protected final ScheduledExecutorService m_scheduler = Executors.newScheduledThreadPool(1);
-    private MapDBMessagesStore m_messageStore;
+    private IMessagesStore m_messageStore;
-    private MapDBSessionsStore m_sessionsStore;
+    private ISessionsStore m_sessionsStore;
 
     public MapDBPersistentStore(IConfig props) {
         this.m_storePath = props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, "");
@@ -71,10 +72,12 @@ public class MapDBPersistentStore {
      *
      * @return the message store instance.
      */
+    @Override
     public IMessagesStore messagesStore() {
         return m_messageStore;
     }
 
+    @Override
     public ISessionsStore sessionsStore() {
         return m_sessionsStore;
     }
@@ -121,6 +124,7 @@ public class MapDBPersistentStore {
         m_sessionsStore.initStore();
     }
 
+    @Override
     public void close() {
         if (this.m_db.isClosed()) {
             LOG.warn("The MapDB store is already closed. Nothing will be done.");

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

@@ -17,6 +17,7 @@
 package io.moquette.spi.persistence;
 
 import io.moquette.spi.ClientSession;
+import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.IMessagesStore.StoredMessage;
 import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.MessageGUID;
@@ -50,9 +51,9 @@ class MapDBSessionsStore implements ISessionsStore {
     private ConcurrentMap<String, Map<Integer, MessageGUID>> m_secondPhaseStore;
 
     private final DB m_db;
-    private final MapDBMessagesStore m_messagesStore;
+    private final IMessagesStore m_messagesStore;
 
-    MapDBSessionsStore(DB db, MapDBMessagesStore messagesStore) {
+    MapDBSessionsStore(DB db, IMessagesStore messagesStore) {
         m_db = db;
         m_messagesStore = messagesStore;
     }
@@ -296,19 +297,6 @@ class MapDBSessionsStore implements ISessionsStore {
         return guid;
     }
 
-    @Override
-    public MessageGUID mapToGuid(String clientID, int messageID) {
-        LOG.debug("Mapping message ID to GUID CId={}, messageId={}", clientID, messageID);
-        ConcurrentMap<Integer, MessageGUID> messageIdToGuid = m_db.getHashMap(messageId2GuidsMapName(clientID));
-        MessageGUID result = messageIdToGuid.get(messageID);
-        LOG.debug("Message ID has been mapped to a GUID CId={}, messageId={}, guid={}", clientID, messageID, result);
-        return result;
-    }
-
-    static String messageId2GuidsMapName(String clientID) {
-        return "guidsMapping_" + clientID;
-    }
-
     @Override
     public StoredMessage getInflightMessage(String clientID, int messageID) {
         LOG.info("Retrieving inflight message CId={}, messageId={}", clientID, messageID);

broker/src/test/java/io/moquette/server/ConfigurationClassLoaderTest.java

@@ -20,6 +20,9 @@ import io.moquette.BrokerConstants;
 import io.moquette.server.config.IConfig;
 import io.moquette.server.config.MemoryConfig;
 import io.moquette.spi.impl.subscriptions.Topic;
+import io.moquette.spi.impl.MemoryMessagesStore;
+import io.moquette.spi.impl.MemoryStorageService;
+import io.moquette.spi.persistence.MemorySessionStore;
 import io.moquette.spi.security.IAuthenticator;
 import io.moquette.spi.security.IAuthorizator;
 import org.junit.After;
@@ -64,6 +67,16 @@ public class ConfigurationClassLoaderTest implements IAuthenticator, IAuthorizat
         assertTrue(true);
     }
 
+    @Test
+    public void loadStorage() throws Exception {
+        Properties props = new Properties(IntegrationUtils.prepareTestProperties());
+        props.setProperty(BrokerConstants.STORAGE_CLASS_NAME, MemoryStorageService.class.getName());
+
+        startServer(props);
+        assertTrue(m_server.getProcessor().getMessagesStore() instanceof MemoryMessagesStore);
+        assertTrue(m_server.getProcessor().getSessionsStore() instanceof MemorySessionStore);
+    }
+
     @Override
     public boolean checkValid(String clientID, String username, byte[] password) {
         return true;

broker/src/test/java/io/moquette/spi/ClientSessionTest.java

@@ -22,7 +22,6 @@ public class ClientSessionTest {
     public void setUp() {
         store = new SubscriptionsStore();
         MemoryStorageService storageService = new MemoryStorageService();
-        storageService.initStore();
         this.sessionsStore = storageService.sessionsStore();
         store.init(sessionsStore);
 

broker/src/test/java/io/moquette/spi/impl/MemoryMessagesStore.java

@@ -34,10 +34,9 @@ public class MemoryMessagesStore implements IMessagesStore {
 
     private Map<Topic, MessageGUID> m_retainedStore = new HashMap<>();
     private Map<MessageGUID, StoredMessage> m_persistentMessageStore = new HashMap<>();
-    private Map<String, Map<Integer, MessageGUID>> m_messageToGuids;
+    private Map<String, Map<Integer, MessageGUID>> m_messageToGuids = new HashMap<>();
 
-    MemoryMessagesStore(Map<String, Map<Integer, MessageGUID>> messageToGuids) {
+    MemoryMessagesStore() {
-        m_messageToGuids = messageToGuids;
     }
 
     @Override
@@ -109,4 +108,11 @@ public class MemoryMessagesStore implements IMessagesStore {
         else
             return messageToGuids.size();
     }
+
+    @Override
+    public MessageGUID mapToGuid(String clientID, int messageID) {
+        HashMap<Integer, MessageGUID> guids = (HashMap<Integer, MessageGUID>) Utils
+                .defaultGet(m_messageToGuids, clientID, new HashMap<Integer, MessageGUID>());
+        return guids.get(messageID);
+    }
 }

broker/src/test/java/io/moquette/spi/impl/MemoryStorageService.java

@@ -18,28 +18,32 @@ package io.moquette.spi.impl;
 
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.ISessionsStore;
-import io.moquette.spi.MessageGUID;
+import io.moquette.spi.IStore;
 import io.moquette.spi.persistence.MemorySessionStore;
-import java.util.HashMap;
-import java.util.Map;
 
-public class MemoryStorageService {
+public class MemoryStorageService implements IStore {
 
     private MemorySessionStore m_sessionsStore;
     private MemoryMessagesStore m_messagesStore;
 
-    public void initStore() {
+    public MemoryStorageService() {
-        Map<String, Map<Integer, MessageGUID>> messageToGuids = new HashMap<>();
+        m_messagesStore = new MemoryMessagesStore();
-        m_messagesStore = new MemoryMessagesStore(messageToGuids);
+        m_sessionsStore = new MemorySessionStore(m_messagesStore);
-        m_sessionsStore = new MemorySessionStore(m_messagesStore, messageToGuids);
+        m_messagesStore.initStore();
+        m_sessionsStore.initStore();
     }
 
+    @Override
     public IMessagesStore messagesStore() {
         return m_messagesStore;
     }
 
+    @Override
     public ISessionsStore sessionsStore() {
         return m_sessionsStore;
     }
 
+    @Override
+    public void close() {
+    }
 }

broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java

@@ -84,7 +84,6 @@ public class ProtocolProcessorTest {
         // sleep to let the messaging batch processor to process the initEvent
         Thread.sleep(300);
         MemoryStorageService memStorage = new MemoryStorageService();
-        memStorage.initStore();
         m_messagesStore = memStorage.messagesStore();
         m_sessionStore = memStorage.sessionsStore();
         // m_messagesStore.initStore();
@@ -128,7 +127,6 @@ public class ProtocolProcessorTest {
 
         // simulate a connect that register a clientID to an IoSession
         MemoryStorageService storageService = new MemoryStorageService();
-        storageService.initStore();
         subs.init(storageService.sessionsStore());
         m_processor.init(
                 subs,
@@ -177,7 +175,6 @@ public class ProtocolProcessorTest {
 
         // simulate a connect that register a clientID to an IoSession
         MemoryStorageService storageService = new MemoryStorageService();
-        storageService.initStore();
         subs.init(storageService.sessionsStore());
         m_processor.init(
                 subs,
@@ -353,7 +350,6 @@ public class ProtocolProcessorTest {
             }
         };
         MemoryStorageService storageService = new MemoryStorageService();
-        storageService.initStore();
         subs.init(storageService.sessionsStore());
 
         // simulate a connect that register a clientID to an IoSession

broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java

@@ -66,7 +66,6 @@ public class ProtocolProcessor_CONNECT_Test {
         // sleep to let the messaging batch processor to process the initEvent
         Thread.sleep(300);
         MemoryStorageService memStorage = new MemoryStorageService();
-        memStorage.initStore();
         m_messagesStore = memStorage.messagesStore();
         m_sessionStore = memStorage.sessionsStore();
         // m_messagesStore.initStore();

broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsStoreTest.java

@@ -43,7 +43,6 @@ public class SubscriptionsStoreTest {
     public void setUp() throws IOException {
         store = new SubscriptionsStore();
         MemoryStorageService storageService = new MemoryStorageService();
-        storageService.initStore();
         this.sessionsStore = storageService.sessionsStore();
         store.init(sessionsStore);
     }
@@ -209,7 +208,6 @@ public class SubscriptionsStoreTest {
         Topic topic = new Topic(t);
         store = new SubscriptionsStore();
         MemoryStorageService memStore = new MemoryStorageService();
-        memStore.initStore();
         ISessionsStore aSessionsStore = memStore.sessionsStore();
         store.init(aSessionsStore);
         Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, MqttQoS.AT_MOST_ONCE);
@@ -223,7 +221,6 @@ public class SubscriptionsStoreTest {
         Topic topic = new Topic(t);
         store = new SubscriptionsStore();
         MemoryStorageService memStore = new MemoryStorageService();
-        memStore.initStore();
         store.init(memStore.sessionsStore());
         Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, MqttQoS.AT_MOST_ONCE);
         sessionsStore.addNewSubscription(sub);
@@ -276,7 +273,6 @@ public class SubscriptionsStoreTest {
     public void removeSubscription_withDifferentClients_subscribedSameTopic() {
         SubscriptionsStore aStore = new SubscriptionsStore();
         MemoryStorageService memStore = new MemoryStorageService();
-        memStore.initStore();
         ISessionsStore sessionsStore = memStore.sessionsStore();
         aStore.init(sessionsStore);
         // subscribe a not active clientID1 to /topic

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

@@ -52,11 +52,9 @@ public class MemorySessionStore implements ISessionsStore {
     // maps clientID->[MessageId -> guid]
     private Map<String, Map<Integer, MessageGUID>> m_secondPhaseStore = new HashMap<>();
 
-    private Map<String, Map<Integer, MessageGUID>> m_messageToGuids;
     private final IMessagesStore m_messagesStore;
 
-    public MemorySessionStore(IMessagesStore messagesStore, Map<String, Map<Integer, MessageGUID>> messageToGuids) {
+    public MemorySessionStore(IMessagesStore messagesStore) {
-        m_messageToGuids = messageToGuids;
         this.m_messagesStore = messagesStore;
     }
 
@@ -235,13 +233,6 @@ public class MemorySessionStore implements ISessionsStore {
         return guid;
     }
 
-    @Override
-    public MessageGUID mapToGuid(String clientID, int messageID) {
-        HashMap<Integer, MessageGUID> guids = (HashMap<Integer, MessageGUID>) Utils
-                .defaultGet(m_messageToGuids, clientID, new HashMap<Integer, MessageGUID>());
-        return guids.get(messageID);
-    }
-
     @Override
     public StoredMessage getInflightMessage(String clientID, int messageID) {
         return null;

perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java

@@ -20,7 +20,6 @@ public class PerformanceTest {
     public static void main(String[] args) throws IOException, InterruptedException {
         SubscriptionsStore store = new SubscriptionsStore();
         MemoryStorageService memStore = new MemoryStorageService();
-        memStore.initStore();
         ISessionsStore aSessionsStore = memStore.sessionsStore();
         store.init(aSessionsStore);