diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 4e2d6559..6580a7bc 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -22,6 +22,7 @@ import java.io.File; */ public class BrokerConstants { public static final String INTERCEPT_HANDLER_PROPERTY_NAME ="intercept.handler"; + public static final String BROKER_INTERCEPTOR_THREAD_POOL_SIZE = "intercept.thread_pool.size"; public static final String PERSISTENT_STORE_PROPERTY_NAME = "persistent_store"; public static final String AUTOSAVE_INTERVAL_PROPERTY_NAME = "autosave_interval"; public static final String PASSWORD_FILE_PROPERTY_NAME = "password_file"; diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index 1d9d7214..a523cf86 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -21,7 +21,9 @@ import java.util.Map; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import io.moquette.BrokerConstants; import io.moquette.interception.InterceptHandler; import io.moquette.interception.Interceptor; import io.moquette.interception.messages.InterceptAcknowledgedMessage; @@ -33,6 +35,7 @@ import io.moquette.interception.messages.InterceptSubscribeMessage; import io.moquette.interception.messages.InterceptUnsubscribeMessage; import io.moquette.parser.proto.messages.ConnectMessage; import io.moquette.parser.proto.messages.PublishMessage; +import io.moquette.server.config.IConfig; import io.moquette.spi.impl.subscriptions.Subscription; /** @@ -44,7 +47,7 @@ final class BrokerInterceptor implements Interceptor { private final Map, List> handlers; private final ExecutorService executor; - BrokerInterceptor(List handlers) { + private BrokerInterceptor(int poolSize, List handlers) { this.handlers = new HashMap<>(); for (Class messageType : InterceptHandler.ALL_MESSAGE_TYPES) { this.handlers.put(messageType, new CopyOnWriteArrayList()); @@ -52,15 +55,37 @@ final class BrokerInterceptor implements Interceptor { for (InterceptHandler handler : handlers) { this.addInterceptHandler(handler); } - executor = Executors.newFixedThreadPool(1); + executor = Executors.newFixedThreadPool(poolSize); + } + + /** + * Configures a broker interceptor, with a thread pool of one thread. + * @param handlers + */ + public BrokerInterceptor(List handlers) { + this(1, handlers); + } + + /** + * Configures a broker interceptor using the pool size specified in the + * IConfig argument. + */ + public BrokerInterceptor(IConfig props, List handlers) { + this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers); } /** * Shutdown graciously the executor service */ - void stop() { - executor.shutdown(); - } + void stop() { + executor.shutdown(); + try { + executor.awaitTermination(10L, TimeUnit.SECONDS); + } catch (InterruptedException e) {} + if (!executor.isTerminated()) { + executor.shutdownNow(); + } + } @Override public void notifyClientConnected(final ConnectMessage msg) { 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 5974611f..3c9e8d6e 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessorBootstrapper.java @@ -102,7 +102,7 @@ public class ProtocolProcessorBootstrapper { LOG.error("Can't load the intercept handler {}", ex); } } - m_interceptor = new BrokerInterceptor(observers); + m_interceptor = new BrokerInterceptor(props, observers); subscriptions.init(m_sessionsStore);