add/remove InterceptHandlers with server running

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

@@ -23,6 +23,7 @@ import io.moquette.spi.impl.subscriptions.Subscription;
 import io.moquette.parser.proto.messages.PublishMessage;
 
 import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -36,7 +37,7 @@ final class BrokerInterceptor implements Interceptor {
     private final ExecutorService executor;
 
     BrokerInterceptor(List<InterceptHandler> handlers) {
-        this.handlers = handlers;
+        this.handlers = new CopyOnWriteArrayList<>(handlers);
         executor = Executors.newFixedThreadPool(1);
     }
 
@@ -118,4 +119,14 @@ final class BrokerInterceptor implements Interceptor {
             });
         }
     }
+
+    @Override
+    public boolean addInterceptHandler(InterceptHandler interceptHandler) {
+        return this.handlers.add(interceptHandler);
+    }
+
+    @Override
+    public boolean removeInterceptHandler(InterceptHandler interceptHandler) {
+        return this.handlers.remove(interceptHandler);
+    }
 }

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

@@ -21,6 +21,7 @@ import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.TimeUnit;
 
+import io.moquette.interception.InterceptHandler;
 import io.moquette.server.ConnectionDescriptor;
 import io.moquette.server.netty.AutoFlushHandler;
 import io.moquette.server.netty.NettyUtils;
@@ -775,4 +776,12 @@ public class ProtocolProcessor {
         }
         channel.flush();
     }
+
+    public boolean addInterceptHandler(InterceptHandler interceptHandler) {
+        return this.m_interceptor.addInterceptHandler(interceptHandler);
+    }
+
+    public boolean removeInterceptHandler(InterceptHandler interceptHandler) {
+        return this.m_interceptor.removeInterceptHandler(interceptHandler);
+    }
 }