From ed9d91e5275fd55e39764e62e7fc51d232e79ac3 Mon Sep 17 00:00:00 2001 From: IODBEEC1 Date: Sat, 9 Jul 2016 09:09:57 +0200 Subject: [PATCH] add/remove InterceptHandlers with server running --- .../io/moquette/interception/Interceptor.java | 4 +++ .../main/java/io/moquette/server/Server.java | 15 +++++++++++ .../moquette/spi/impl/BrokerInterceptor.java | 13 ++++++++- .../moquette/spi/impl/ProtocolProcessor.java | 9 +++++++ .../spi/impl/BrokerInterceptorTest.java | 27 +++++++++++++++++++ 5 files changed, 67 insertions(+), 1 deletion(-) diff --git a/broker/src/main/java/io/moquette/interception/Interceptor.java b/broker/src/main/java/io/moquette/interception/Interceptor.java index be7e4ed1..634b1748 100644 --- a/broker/src/main/java/io/moquette/interception/Interceptor.java +++ b/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); } diff --git a/broker/src/main/java/io/moquette/server/Server.java b/broker/src/main/java/io/moquette/server/Server.java index 92538f39..b53bdf8b 100644 --- a/broker/src/main/java/io/moquette/server/Server.java +++ b/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); + } + } 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 a7a42cd1..81899281 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/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 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); + } } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 7c484f58..69396dbc 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/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); + } } diff --git a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java index 8a65fa4f..b84b4896 100644 --- a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java +++ b/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"))); + } }