Make the size of the BrokerInterceptor thread pool configurable.

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";

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<Class<?>, List<InterceptHandler>> handlers;
     private final ExecutorService executor;
 
-	BrokerInterceptor(List<InterceptHandler> handlers) {
+	private BrokerInterceptor(int poolSize, List<InterceptHandler> handlers) {
 		this.handlers = new HashMap<>();
 		for (Class<?> messageType : InterceptHandler.ALL_MESSAGE_TYPES) {
 			this.handlers.put(messageType, new CopyOnWriteArrayList<InterceptHandler>());
@@ -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<InterceptHandler> handlers) {
+		this(1, handlers);
+	}
+	
+	/**
+	 * Configures a broker interceptor using the pool size specified in the
+	 * IConfig argument.
+	 */
+	public BrokerInterceptor(IConfig props, List<InterceptHandler> 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) {

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);