From dc55ff4078cb467fe795a697eac8c417e7869811 Mon Sep 17 00:00:00 2001 From: andrea Date: Sat, 8 Aug 2015 18:37:32 +0200 Subject: [PATCH] Removed never used and unsufull no-op interceptor, forced usage of BrokerInterceptor, changed BrokerInterceptor to work on a list of InterceptHandler --- .../moquette/interception/Interceptor.java | 6 +- .../moquette/spi/impl/BrokerInterceptor.java | 77 +++++++++++-------- .../moquette/spi/impl/ProtocolProcessor.java | 32 +++----- .../moquette/spi/impl/SimpleMessaging.java | 30 +++++--- .../subscriptions/SubscriptionsStore.java | 1 - .../spi/impl/BrokerInterceptorTest.java | 7 +- .../spi/impl/ProtocolProcessorTest.java | 27 ++++--- 7 files changed, 94 insertions(+), 86 deletions(-) diff --git a/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java b/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java index 26c0021c..c19563ee 100644 --- a/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java +++ b/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java @@ -23,8 +23,7 @@ import org.eclipse.moquette.spi.impl.subscriptions.Subscription; * This interface is to be used internally by the broker components. *

* An interface is used instead of a class to allow more flexibility in changing - * an implementation. For instance, this interface provide a lightweight - * no-handler interceptor. + * an implementation. *

* Interceptor implementations forward notifications to a InterceptHandler, * that is normally a field. So, the implementations should act as a proxy to a custom @@ -34,9 +33,6 @@ import org.eclipse.moquette.spi.impl.subscriptions.Subscription; * @author Wagner Macedo */ public interface Interceptor { - /** - * When no custom handler is provided, this interceptor should be used. - */ void notifyClientConnected(ConnectMessage msg); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java index de393e87..7786b30f 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java @@ -21,6 +21,7 @@ import org.eclipse.moquette.proto.messages.ConnectMessage; import org.eclipse.moquette.proto.messages.PublishMessage; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; +import java.util.List; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -30,11 +31,11 @@ import java.util.concurrent.Executors; * @author Wagner Macedo */ final class BrokerInterceptor implements Interceptor { - private final InterceptHandler handler; + private final List handlers; private final ExecutorService executor; - BrokerInterceptor(InterceptHandler handler) { - this.handler = handler; + BrokerInterceptor(List handlers) { + this.handlers = handlers; executor = Executors.newFixedThreadPool(1); } @@ -47,51 +48,61 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientConnected(final ConnectMessage msg) { - executor.execute(new Runnable() { - @Override - public void run() { - handler.onConnect(msg.readOnlyClone()); - } - }); + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onConnect(msg.readOnlyClone()); + } + }); + } } @Override public void notifyClientDisconnected(final String clientID) { - executor.execute(new Runnable() { - @Override - public void run() { - handler.onDisconnect(clientID); - } - }); + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onDisconnect(clientID); + } + }); + } } @Override public void notifyTopicPublished(final PublishMessage msg) { - executor.execute(new Runnable() { - @Override - public void run() { - handler.onPublish(msg.readOnlyClone()); - } - }); + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onPublish(msg.readOnlyClone()); + } + }); + } } @Override public void notifyTopicSubscribed(final Subscription sub) { - executor.execute(new Runnable() { - @Override - public void run() { - handler.onSubscribe(sub.clone()); - } - }); + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onSubscribe(sub); + } + }); + } } @Override public void notifyTopicUnsubscribed(final String sub) { - executor.execute(new Runnable() { - @Override - public void run() { - handler.onUnsubscribe(sub); - } - }); + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onUnsubscribe(sub); + } + }); + } } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 3dc73f29..a736925c 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java @@ -23,9 +23,6 @@ import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import io.netty.util.internal.StringUtil; -import org.apache.commons.codec.binary.StringUtils; -import org.eclipse.moquette.interception.Interceptor; import org.eclipse.moquette.server.netty.NettyChannel; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; @@ -106,7 +103,7 @@ class ProtocolProcessor implements EventHandler { private IMessagesStore m_messagesStore; private ISessionsStore m_sessionsStore; private IAuthenticator m_authenticator; - private Interceptor interceptor = null; + private BrokerInterceptor m_interceptor; //maps clientID to Will testament, if specified on CONNECT private Map m_willStore = new HashMap<>(); @@ -116,13 +113,6 @@ class ProtocolProcessor implements EventHandler { ProtocolProcessor() {} - void init(SubscriptionsStore subscriptions, IMessagesStore storageService, - ISessionsStore sessionsStore, - IAuthenticator authenticator, - boolean allowAnonymous, IAuthorizator authorizator) { - init(subscriptions, storageService, sessionsStore, authenticator, allowAnonymous, authorizator, null); - } - /** * @param subscriptions the subscription store where are stored all the existing * clients subscriptions. @@ -137,11 +127,9 @@ class ProtocolProcessor implements EventHandler { void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, - boolean allowAnonymous, IAuthorizator authorizator, Interceptor interceptor) { + boolean allowAnonymous, IAuthorizator authorizator, BrokerInterceptor interceptor) { //m_clientIDs = clientIDs; - if (interceptor != null) { - this.interceptor = interceptor; - } + this.m_interceptor = interceptor; this.subscriptions = subscriptions; this.allowAnonymous = allowAnonymous; m_authorizator = authorizator; @@ -177,7 +165,7 @@ class ProtocolProcessor implements EventHandler { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); session.write(okResp); - interceptor.notifyClientConnected(msg); + m_interceptor.notifyClientConnected(msg); return; } @@ -233,7 +221,7 @@ class ProtocolProcessor implements EventHandler { AbstractMessage.QOSType willQos = AbstractMessage.QOSType.values()[msg.getWillQos()]; byte[] willPayload = msg.getWillMessage().getBytes(); ByteBuffer bb = (ByteBuffer) ByteBuffer.allocate(willPayload.length).put(willPayload).flip(); - //save the will testment in the clientID store + //save the will testament in the clientID store WillMessage will = new WillMessage(msg.getWillTopic(), bb, msg.isWillRetain(),willQos ); m_willStore.put(msg.getClientID(), will); } @@ -253,7 +241,7 @@ class ProtocolProcessor implements EventHandler { okResp.setSessionPresent(true); } session.write(okResp); - interceptor.notifyClientConnected(msg); + m_interceptor.notifyClientConnected(msg); LOG.info("Create persistent session for clientID <{}>", msg.getClientID()); m_sessionsStore.addNewSubscription(Subscription.createEmptySubscription(msg.getClientID(), true)); //null means EmptySubscription @@ -317,7 +305,7 @@ class ProtocolProcessor implements EventHandler { String user = (String) session.getAttribute(NettyChannel.ATTR_KEY_USERNAME); if (m_authorizator.canWrite(topic, user, clientID)) { executePublish(clientID, msg); - interceptor.notifyTopicPublished(msg); + m_interceptor.notifyTopicPublished(msg); } else { LOG.debug("topic {} doesn't have write credentials", topic); } @@ -576,7 +564,7 @@ class ProtocolProcessor implements EventHandler { m_willStore.remove(clientID); LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession); - interceptor.notifyClientDisconnected(clientID); + m_interceptor.notifyClientDisconnected(clientID); } void processConnectionLost(LostConnectionEvent evt) { @@ -610,7 +598,7 @@ class ProtocolProcessor implements EventHandler { for (String topic : topics) { subscriptions.removeSubscription(topic, clientID); m_sessionsStore.removeSubscription(topic, clientID); - interceptor.notifyTopicUnsubscribed(topic); + m_interceptor.notifyTopicUnsubscribed(topic); } //ack the client @@ -655,7 +643,7 @@ class ProtocolProcessor implements EventHandler { subscriptions.add(newSubscription); //notify the Observables - interceptor.notifyTopicSubscribed(newSubscription); + m_interceptor.notifyTopicSubscribed(newSubscription); //scans retained messages to be published to the new subscription Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java index 8ad92961..e19de34b 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java @@ -18,9 +18,9 @@ package org.eclipse.moquette.spi.impl; import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.dsl.Disruptor; +import io.netty.util.internal.StringUtil; import org.HdrHistogram.Histogram; import org.eclipse.moquette.interception.InterceptHandler; -import org.eclipse.moquette.interception.Interceptor; import org.eclipse.moquette.proto.messages.AbstractMessage; import org.eclipse.moquette.spi.impl.security.*; import org.eclipse.moquette.server.ServerChannel; @@ -40,6 +40,8 @@ import java.io.File; import java.lang.reflect.InvocationTargetException; import java.lang.reflect.Method; import java.text.ParseException; +import java.util.ArrayList; +import java.util.List; import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -75,7 +77,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { private ExecutorService m_executor; private Disruptor m_disruptor; - private Interceptor interceptor; + private BrokerInterceptor m_interceptor; private static SimpleMessaging INSTANCE; @@ -194,14 +196,18 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_sessionsStore = mapStorage; m_storageService.initStore(); - - try { - final InterceptHandler handler = Class.forName(props.getProperty("intercept.handler")) - .asSubclass(InterceptHandler.class).newInstance(); - interceptor = new BrokerInterceptor(handler); - } catch (Throwable ex) { - LOG.error("Can't load the intercept handler {}", ex); + + List observers = new ArrayList<>(); + String interceptorClassName = props.getProperty("intercept.handler"); + if (interceptorClassName != null && !interceptorClassName.isEmpty()) { + try { + InterceptHandler handler = Class.forName(interceptorClassName).asSubclass(InterceptHandler.class).newInstance(); + observers.add(handler); + } catch (Throwable ex) { + LOG.error("Can't load the intercept handler {}", ex); + } } + m_interceptor = new BrokerInterceptor(observers); //List storedSubscriptions = m_sessionsStore.listAllSubscriptions(); //subscriptions.init(storedSubscriptions); @@ -251,7 +257,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(ALLOW_ANONYMOUS_PROPERTY_NAME, "true")); - m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator, allowAnonymous, authorizator, interceptor); + m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor); } private Object loadClass(String className, Class cls) { @@ -291,8 +297,8 @@ public class SimpleMessaging implements IMessaging, EventHandler { private void processStop() { LOG.debug("processStop invoked"); - if (interceptor != null) { - ((BrokerInterceptor) interceptor).stop(); + if (m_interceptor != null) { + this.m_interceptor.stop(); } m_storageService.close(); LOG.debug("subscription tree {}", subscriptions.dumpTree()); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java index ab210678..95a03d6d 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java @@ -19,7 +19,6 @@ import java.text.ParseException; import java.util.*; import java.util.concurrent.LinkedBlockingDeque; -import org.eclipse.moquette.interception.Interceptor; import org.eclipse.moquette.spi.ISessionsStore; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/BrokerInterceptorTest.java index 11c3a712..79b71a5a 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/BrokerInterceptorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/BrokerInterceptorTest.java @@ -24,6 +24,7 @@ import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; +import java.util.Arrays; import java.util.concurrent.atomic.AtomicInteger; import static org.junit.Assert.*; @@ -36,7 +37,7 @@ public class BrokerInterceptorTest { private static final AtomicInteger n = new AtomicInteger(0); // Interceptor loaded with a custom InterceptHandler special for the tests - private static final BrokerInterceptor interceptor = new BrokerInterceptor(new InterceptHandler() { + private static final class MockObserver implements InterceptHandler { @Override public void onConnect(ConnectMessage msg) { n.set(40); @@ -61,7 +62,9 @@ public class BrokerInterceptorTest { public void onUnsubscribe(String topic) { n.set(80); } - }); + } + + private static final BrokerInterceptor interceptor = new BrokerInterceptor(Arrays.asList(new MockObserver())); @BeforeClass public static void beforeAllTests() { diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 1e3d2bf0..678ecf1d 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -21,6 +21,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import io.netty.util.AttributeKey; +import org.eclipse.moquette.interception.InterceptHandler; import org.eclipse.moquette.server.netty.NettyChannel; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; @@ -59,6 +60,9 @@ public class ProtocolProcessorTest { final static String TEST_PWD = "fakepwd"; final static String EVIL_TEST_USER = "eviluser"; final static String EVIL_TEST_PWD = "unsecret"; + + final static List EMPTY_OBSERVERS = Collections.emptyList(); + final static BrokerInterceptor NO_OBSERVERS_INTERCEPTOR = new BrokerInterceptor(EMPTY_OBSERVERS); DummyChannel m_session; byte m_returnCode; @@ -172,7 +176,8 @@ public class ProtocolProcessorTest { subscriptions = new SubscriptionsStore(); subscriptions.init(new MemoryStorageService()); m_processor = new ProtocolProcessor(); - m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator()); + m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, + new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); } @Test @@ -263,7 +268,7 @@ public class ProtocolProcessorTest { @Test public void prohibitAnonymousClient() { connMsg.setClientID("123"); - m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator()); + m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); //Exercise m_processor.processConnect(m_session, connMsg); @@ -277,7 +282,7 @@ public class ProtocolProcessorTest { connMsg.setClientID("123"); connMsg.setUserFlag(true); connMsg.setUsername(TEST_USER + "_fake"); - m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator()); + m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, false, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); //Exercise m_processor.processConnect(m_session, connMsg); @@ -289,7 +294,7 @@ public class ProtocolProcessorTest { @Test public void acceptAnonymousClient() { connMsg.setClientID("123"); - m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator()); + m_processor.init(subscriptions, m_storageService, m_sessionStore, m_mockAuthenticator, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); //Exercise m_processor.processConnect(m_session, connMsg); @@ -376,7 +381,7 @@ public class ProtocolProcessorTest { //simulate a connect that register a clientID to an IoSession subs.init(new MemoryStorageService()); - m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator()); + m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setProcotolVersion((byte)3); connectMessage.setClientID(FAKE_CLIENT_ID); @@ -420,7 +425,7 @@ public class ProtocolProcessorTest { //simulate a connect that register a clientID to an IoSession subs.init(new MemoryStorageService()); - m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator()); + m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); MockReceiverChannel firstReceiverSession = new MockReceiverChannel(); ConnectMessage connectMessage = new ConnectMessage(); @@ -559,7 +564,7 @@ public class ProtocolProcessorTest { subs.init(new MemoryStorageService()); //simulate a connect that register a clientID to an IoSession - m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator()); + m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setClientID(FAKE_PUBLISHER_ID); connectMessage.setProcotolVersion((byte)3); @@ -598,7 +603,7 @@ public class ProtocolProcessorTest { ByteBuffer.wrap("Hello".getBytes()), true, FAKE_PUBLISHER_ID, 120); m_storageService.storePublishForFuture(retainedMessage); - m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator()); + m_processor.init(subs, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); ConnectMessage connectMessage = new ConnectMessage(); connectMessage.setClientID(FAKE_PUBLISHER_ID); connectMessage.setProcotolVersion((byte)3); @@ -617,7 +622,7 @@ public class ProtocolProcessorTest { List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator()); + m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); @@ -642,7 +647,7 @@ public class ProtocolProcessorTest { List inactiveSubscriptions = Arrays.asList(inactiveSub); when(mockedSubscriptions.matches(eq("/topic"))).thenReturn(inactiveSubscriptions); m_processor = new ProtocolProcessor(); - m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator()); + m_processor.init(mockedSubscriptions, m_storageService, m_sessionStore, null, true, new PermitAllAuthorizator(), NO_OBSERVERS_INTERCEPTOR); //Exercise ByteBuffer buffer = ByteBuffer.allocate(5).put("Hello".getBytes()); @@ -718,7 +723,7 @@ public class ProtocolProcessorTest { publishedForwarded.add(new PublishEvent(topic, qos, message, retained, clientId, messageID)); } }; - processor.init(subscriptions, memoryMessageStore, null, null, true, null); + processor.init(subscriptions, memoryMessageStore, null, null, true, null, NO_OBSERVERS_INTERCEPTOR); //Exercise processor.forward2Subscribers(forwardPublish);