From 16a9a0ccd07d1d09a36e2bc48cb20d9da5c1d239 Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 18 Jun 2017 11:01:25 +0200 Subject: [PATCH] Fixed runtime loading of persistence implementation --- .../persistence/MemoryStorageService.java | 5 +- .../server/netty/NettyMQTTHandler.java | 8 +- .../impl/ProtocolProcessorBootstrapper.java | 131 ++++++++---------- .../io/moquette/spi/ClientSessionTest.java | 2 +- .../spi/impl/ProtocolProcessorTest.java | 8 +- .../impl/ProtocolProcessor_CONNECT_Test.java | 2 +- .../SubscriptionsDirectoryTest.java | 8 +- .../impl/subscriptions/PerformanceTest.java | 2 +- 8 files changed, 77 insertions(+), 89 deletions(-) diff --git a/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java b/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java index 05b37a52..0f1e7b0e 100644 --- a/broker/src/main/java/io/moquette/persistence/MemoryStorageService.java +++ b/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(); diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index 63c23146..52c170ad 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/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( - "An unexpected exception was caught while processing MQTT message. " - + "Closing Netty channel. MqttClientId = {}, cause = {}, errorMessage = {}.", - NettyUtils.clientID(ctx.channel()), - cause.getCause(), - cause.getMessage()); + LOG.error("An unexpected exception was caught while processing MQTT message. Closing Netty channel. CId={}, " + + "cause={}, errorMessage={}", NettyUtils.clientID(ctx.channel()), cause.getCause(), cause.getMessage()); ctx.close(); } 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 2cc7c892..6531b7ab 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/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 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 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 loadClass(String className, Class intrface, Class constructorArgClass, U props) { T instance = null; try { - LOG.info("Loading class. ClassName={}, interfaceName={}", className, intrface.getName()); - Class clazz = Class.forName(className); - - // check if method getInstance exists - Method method = clazz.getMethod("getInstance", new Class[]{}); + // 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(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( - "Invoking getInstance() method. ClassName = {}, interfaceName = {}.", - className, - 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()); + LOG.info("Invoking default constructor. ClassName={}, interfaceName={}", + className, intrface.getName()); + // fallback to default constructor instance = this.getClass().getClassLoader() .loadClass(className) .asSubclass(intrface) - .getConstructor(constructorArgClass) - .newInstance(props); + .newInstance(); } 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()); + LOG.error("Unable to invoke default constructor. ClassName={}, interfaceName={}, cause={}, errorMessage={}", + 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; diff --git a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java b/broker/src/test/java/io/moquette/spi/ClientSessionTest.java index c2be220c..aa43eb92 100644 --- a/broker/src/test/java/io/moquette/spi/ClientSessionTest.java +++ b/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); diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java index 9053bdac..02eae29e 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessorTest.java +++ b/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 diff --git a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java b/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java index 011d3b15..0c24b929 100644 --- a/broker/src/test/java/io/moquette/spi/impl/ProtocolProcessor_CONNECT_Test.java +++ b/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(); diff --git a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectoryTest.java b/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectoryTest.java index c6af98e6..f5eb82c2 100644 --- a/broker/src/test/java/io/moquette/spi/impl/subscriptions/SubscriptionsDirectoryTest.java +++ b/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 diff --git a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java b/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java index ce7d18db..bec33620 100644 --- a/perf/src/main/java/io/moquette/spi/impl/subscriptions/PerformanceTest.java +++ b/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);