Merge pull request #187 from iodb/master

add/remove InterceptHandlers with server running

broker/src/main/java/io/moquette/interception/Interceptor.java

@@ -46,4 +46,8 @@ public interface Interceptor {
     void notifyTopicUnsubscribed(String topic, String clientID, final String username);
 
     void notifyMessageAcknowledged(InterceptAcknowledgedMessage msg);
+
+    boolean addInterceptHandler(InterceptHandler interceptHandler);
+
+    boolean removeInterceptHandler(InterceptHandler interceptHandler);
 }

broker/src/main/java/io/moquette/server/Server.java

@@ -155,4 +155,19 @@ public class Server {
         m_initialized = false;
         LOG.info("Server stopped");
     }
+
+    public boolean addInterceptHandler(InterceptHandler interceptHandler) {
+        if (!m_initialized) {
+            throw new IllegalStateException("Can't register interceptors on a server that is not yet started");
+        }
+        return m_processor.addInterceptHandler(interceptHandler);
+    }
+
+    public boolean removeInterceptHandler(InterceptHandler interceptHandler) {
+        if (!m_initialized) {
+            throw new IllegalStateException("Can't deregister interceptors from a server that is not yet started");
+        }
+        return m_processor.removeInterceptHandler(interceptHandler);
+    }
+
 }

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;
@@ -786,4 +787,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);
+    }
 }

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

@@ -29,6 +29,8 @@ import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.Assert.*;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
 
 /**
  * @author Wagner Macedo
@@ -122,4 +124,29 @@ public class BrokerInterceptorTest {
         interval();
         assertEquals(80, n.get());
     }
+
+    @Test
+    public void testAddAndRemoveInterceptHandler() throws Exception {
+        InterceptHandler interceptHandlerMock1 = mock(InterceptHandler.class);
+        InterceptHandler interceptHandlerMock2 = mock(InterceptHandler.class);
+        // add
+        interceptor.addInterceptHandler(interceptHandlerMock1);
+        interceptor.addInterceptHandler(interceptHandlerMock2);
+
+        Subscription subscription = new Subscription("cli1", "o2", AbstractMessage.QOSType.MOST_ONE);
+        interceptor.notifyTopicSubscribed(subscription, "cli1234");
+        interval();
+
+        verify(interceptHandlerMock1).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234")));
+        verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1234")));
+
+        // remove
+        interceptor.removeInterceptHandler(interceptHandlerMock1);
+
+        interceptor.notifyTopicSubscribed(subscription, "cli1235");
+        interval();
+
+        verifyNoMoreInteractions(interceptHandlerMock1);
+        verify(interceptHandlerMock2).onSubscribe(refEq(new InterceptSubscribeMessage(subscription, "cli1235")));
+    }
 }