Fixed runtime loading of persistence implementation

broker/src/main/java/io/moquette/persistence/MemoryStorageService.java

@@ -16,16 +16,19 @@
 
 package io.moquette.persistence;
 
+import io.moquette.server.config.IConfig;
 import io.moquette.spi.IMessagesStore;
 import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.IStore;
 
+import java.util.concurrent.ScheduledExecutorService;
+
 public class MemoryStorageService implements IStore {
 
     private MemorySessionStore m_sessionsStore;
     private MemoryMessagesStore m_messagesStore;
 
-    public MemoryStorageService() {
+    public MemoryStorageService(IConfig props, ScheduledExecutorService scheduler) {
         m_messagesStore = new MemoryMessagesStore();
         m_sessionsStore = new MemorySessionStore();
         m_messagesStore.initStore();

broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java

@@ -102,12 +102,8 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
-        LOG.error(
+        LOG.error("An unexpected exception was caught while processing MQTT message. Closing Netty channel. CId={}, " +
-                "An unexpected exception was caught while processing MQTT message. "
+            "cause={}, errorMessage={}", NettyUtils.clientID(ctx.channel()), cause.getCause(), cause.getMessage());
-                + "Closing Netty channel. MqttClientId = {}, cause = {}, errorMessage = {}.",
-                NettyUtils.clientID(ctx.channel()),
-                cause.getCause(),
-                cause.getMessage());
         ctx.close();
     }
 

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

@@ -34,11 +34,12 @@ import io.moquette.spi.security.IAuthorizator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
 import java.text.ParseException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
 
 /**
  * It's main responsibility is bootstrap the ProtocolProcessor.
@@ -46,7 +47,7 @@ import java.util.List;
 public class ProtocolProcessorBootstrapper {
 
     private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessorBootstrapper.class);
-    public static final String MAPDB_STORE_CLASS = "io.moquette.persistence.MemoryStorageService";
+    public static final String INMEMDB_STORE_CLASS = "io.moquette.persistence.MemoryStorageService";
 
     private ISessionsStore m_sessionsStore;
 
@@ -83,12 +84,16 @@ public class ProtocolProcessorBootstrapper {
             IAuthenticator authenticator, IAuthorizator authorizator, Server server) {
         IMessagesStore messagesStore;
         LOG.info("Initializing messages and sessions stores...");
-        String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, MAPDB_STORE_CLASS);
+        String storageClassName = props.getProperty(BrokerConstants.STORAGE_CLASS_NAME, INMEMDB_STORE_CLASS);
         if (storageClassName == null || storageClassName.isEmpty()) {
             LOG.error("storage_class property not defined");
-            throw new IllegalArgumentException("Can't start a valid persistence layer");
+            throw new IllegalArgumentException("Can't find a valid persistence layer");
         }
-        final IStore store = loadClass(storageClassName, IStore.class, Server.class, server);
+        final IStore store = instantiateConfiguredStore(storageClassName, props, server.getScheduler());
+        if (store == null) {
+            throw new IllegalArgumentException("Can't start the persistence layer");
+        }
+        store.initStore();
         messagesStore = store.messagesStore();
         m_sessionsStore = store.sessionsStore();
         this.subscriptionsStore = m_sessionsStore.subscriptionStore();
@@ -170,83 +175,67 @@ public class ProtocolProcessorBootstrapper {
         return m_processor;
     }
 
+    private IStore instantiateConfiguredStore(String storageClassName, IConfig props,
+                                              ScheduledExecutorService scheduledExecutor) {
+        LOG.info("Loading storage class {}", storageClassName);
+        Class<? extends IStore> storageClass;
+        try {
+            storageClass = this.getClass().getClassLoader()
+                .loadClass(storageClassName)
+                .asSubclass(IStore.class);
+        } catch (ClassNotFoundException cnfex) {
+            LOG.error("Cannot find storage class " + storageClassName + " in classpath", cnfex);
+            return null;
+        }
+
+        final Constructor<? extends IStore> constructor;
+        try {
+            constructor = storageClass
+                .getConstructor(IConfig.class, ScheduledExecutorService.class);
+        } catch (NoSuchMethodException nsmex) {
+            LOG.error("Cannot find constructor with required params IConfig, ScheduledExecutorService ", nsmex);
+            return null;
+        }
+        try {
+            return constructor.newInstance(props, scheduledExecutor);
+        } catch (InstantiationException | IllegalAccessException | InvocationTargetException ex) {
+            LOG.error("Cannot instantiate the " + storageClassName + " instance", ex);
+        }
+
+        return null;
+    }
+
     @SuppressWarnings("unchecked")
     private <T, U> T loadClass(String className, Class<T> intrface, Class<U> constructorArgClass, U props) {
         T instance = null;
         try {
-            LOG.info("Loading class. ClassName={}, interfaceName={}", className, intrface.getName());
+            // check if constructor with constructor arg class parameter
-            Class<?> clazz = Class.forName(className);
+            // exists
-
+            LOG.info("Invoking constructor with {} argument. ClassName={}, interfaceName={}",
-            // check if method getInstance exists
+                    constructorArgClass.getName(), className, intrface.getName());
-            Method method = clazz.getMethod("getInstance", new Class[]{});
+            instance = this.getClass().getClassLoader()
+                .loadClass(className)
+                .asSubclass(intrface)
+                .getConstructor(constructorArgClass)
+                .newInstance(props);
+        } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
+            LOG.warn("Unable to invoke constructor with {} argument. ClassName={}, interfaceName={}, cause={}, errorMessage={}",
+                    constructorArgClass.getName(), className, intrface.getName(), ex.getCause(), ex.getMessage());
+            return null;
+        } catch (NoSuchMethodException | InvocationTargetException e) {
             try {
-                LOG.info(
+                LOG.info("Invoking default constructor. ClassName={}, interfaceName={}",
-                        "Invoking getInstance() method. ClassName = {}, interfaceName = {}.",
+                        className, intrface.getName());
-                        className,
+                // fallback to default constructor
-                        intrface.getName());
-                instance = (T) method.invoke(null, new Object[]{});
-            } catch (IllegalArgumentException | InvocationTargetException | IllegalAccessException ex) {
-                LOG.error(
-                        "Unable to invoke getInstance() method. ClassName = {}, interfaceName = {}, cause = {}, "
-                        + "errorMessage = {}.",
-                        className,
-                        intrface.getName(),
-                        ex.getCause(),
-                        ex.getMessage());
-                return null;
-            }
-        } catch (NoSuchMethodException nsmex) {
-            try {
-                // check if constructor with constructor arg class parameter
-                // exists
-                LOG.info("Invoking constructor with {} argument. ClassName = {}, interfaceName = {}.",
-                        constructorArgClass.getName(), className, intrface.getName());
                 instance = this.getClass().getClassLoader()
                     .loadClass(className)
                     .asSubclass(intrface)
-                    .getConstructor(constructorArgClass)
+                    .newInstance();
-                    .newInstance(props);
             } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
-                LOG.warn(
+                LOG.error("Unable to invoke default constructor. ClassName={}, interfaceName={}, cause={}, errorMessage={}",
-                        "Unable to invoke constructor with {} argument. ClassName = {}, interfaceName = {}, cause = {}"
+                        className, intrface.getName(), ex.getCause(), ex.getMessage());
-                                + ", errorMessage = {}.",
-                        constructorArgClass.getName(),
-                        className,
-                        intrface.getName(),
-                        ex.getCause(),
-                        ex.getMessage());
                 return null;
-            } catch (NoSuchMethodException | InvocationTargetException e) {
-                try {
-                    LOG.info(
-                            "Invoking default constructor. ClassName = {}, interfaceName = {}.",
-                            className,
-                            intrface.getName());
-                    // fallback to default constructor
-                    instance = this.getClass().getClassLoader().loadClass(className).asSubclass(intrface).newInstance();
-                } catch (InstantiationException | IllegalAccessException | ClassNotFoundException ex) {
-                    LOG.error(
-                            "Unable to invoke default constructor. ClassName = {}, interfaceName = {}, cause = {}, "
-                            + "errorMessage = {}.",
-                            className,
-                            intrface.getName(),
-                            ex.getCause(),
-                            ex.getMessage());
-                    return null;
-                }
             }
-        } catch (ClassNotFoundException ex) {
-            LOG.error("The class does not exist. ClassName = {}, interfaceName = {}.", className, intrface.getName());
-            return null;
-        } catch (SecurityException ex) {
-            LOG.error(
-                    "Unable to load class due to a security violation. ClassName = {}, interfaceName = {}, cause = {}, "
-                    + "errorMessage = {}.",
-                    className,
-                    intrface.getName(),
-                    ex.getCause(),
-                    ex.getMessage());
-            return null;
         }
 
         return instance;

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

@@ -36,7 +36,7 @@ public class ClientSessionTest {
     @Before
     public void setUp() {
         store = new SubscriptionsDirectory();
-        MemoryStorageService storageService = new MemoryStorageService();
+        MemoryStorageService storageService = new MemoryStorageService(null, null);
         this.sessionsStore = storageService.sessionsStore();
         store.init(sessionsStore);
 

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

@@ -80,7 +80,7 @@ public class ProtocolProcessorTest {
 
         // sleep to let the messaging batch processor to process the initEvent
         Thread.sleep(300);
-        MemoryStorageService memStorage = new MemoryStorageService();
+        MemoryStorageService memStorage = new MemoryStorageService(null, null);
         m_messagesStore = memStorage.messagesStore();
         m_sessionStore = memStorage.sessionsStore();
         // m_messagesStore.initStore();
@@ -117,7 +117,7 @@ public class ProtocolProcessorTest {
         };
 
         // simulate a connect that register a clientID to an IoSession
-        MemoryStorageService storageService = new MemoryStorageService();
+        MemoryStorageService storageService = new MemoryStorageService(null, null);
         subs.init(storageService.sessionsStore());
         m_processor.init(
                 subs,
@@ -165,7 +165,7 @@ public class ProtocolProcessorTest {
         };
 
         // simulate a connect that register a clientID to an IoSession
-        MemoryStorageService storageService = new MemoryStorageService();
+        MemoryStorageService storageService = new MemoryStorageService(null, null);
         subs.init(storageService.sessionsStore());
         m_processor.init(subs, m_messagesStore, m_sessionStore, null, true, new PermitAllAuthorizator(),
                 NO_OBSERVERS_INTERCEPTOR);
@@ -322,7 +322,7 @@ public class ProtocolProcessorTest {
                 }
             }
         };
-        MemoryStorageService storageService = new MemoryStorageService();
+        MemoryStorageService storageService = new MemoryStorageService(null, null);
         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

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

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

@@ -42,7 +42,7 @@ public class SubscriptionsDirectoryTest {
     @Before
     public void setUp() throws IOException {
         store = new SubscriptionsDirectory();
-        MemoryStorageService storageService = new MemoryStorageService();
+        MemoryStorageService storageService = new MemoryStorageService(null, null);
         this.sessionsStore = storageService.sessionsStore();
         store.init(sessionsStore);
 
@@ -212,7 +212,7 @@ public class SubscriptionsDirectoryTest {
         Topic subscription = new Topic(s);
         Topic topic = new Topic(t);
         store = new SubscriptionsDirectory();
-        MemoryStorageService memStore = new MemoryStorageService();
+        MemoryStorageService memStore = new MemoryStorageService(null, null);
         ISessionsStore aSessionsStore = memStore.sessionsStore();
         aSessionsStore.createNewSession("FAKE_CLI_ID_1", false);
         store.init(aSessionsStore);
@@ -226,7 +226,7 @@ public class SubscriptionsDirectoryTest {
         Topic subscription = new Topic(s);
         Topic topic = new Topic(t);
         store = new SubscriptionsDirectory();
-        MemoryStorageService memStore = new MemoryStorageService();
+        MemoryStorageService memStore = new MemoryStorageService(null, null);
         store.init(memStore.sessionsStore());
         Subscription sub = new Subscription("FAKE_CLI_ID_1", subscription, MqttQoS.AT_MOST_ONCE);
         this.subscriptionsStore.addNewSubscription(sub);
@@ -278,7 +278,7 @@ public class SubscriptionsDirectoryTest {
     @Test
     public void removeSubscription_withDifferentClients_subscribedSameTopic() {
         SubscriptionsDirectory aStore = new SubscriptionsDirectory();
-        MemoryStorageService memStore = new MemoryStorageService();
+        MemoryStorageService memStore = new MemoryStorageService(null, null);
         ISessionsStore sessionsStore = memStore.sessionsStore();
         aStore.init(sessionsStore);
         // subscribe a not active clientID1 to /topic

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

@@ -35,7 +35,7 @@ public final class PerformanceTest {
      */
     public static void main(String[] args) throws IOException, InterruptedException {
         SubscriptionsDirectory store = new SubscriptionsDirectory();
-        MemoryStorageService memStore = new MemoryStorageService();
+        MemoryStorageService memStore = new MemoryStorageService(null, null);
         ISessionsStore aSessionsStore = memStore.sessionsStore();
         store.init(aSessionsStore);