diff --git a/broker/src/main/java/org/eclipse/moquette/interception/DefaultInterceptHandler.java b/broker/src/main/java/org/eclipse/moquette/interception/DefaultInterceptHandler.java new file mode 100644 index 00000000..44cbd748 --- /dev/null +++ b/broker/src/main/java/org/eclipse/moquette/interception/DefaultInterceptHandler.java @@ -0,0 +1,47 @@ +/* + * Copyright (c) 2012-2014 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package org.eclipse.moquette.interception; + +import org.eclipse.moquette.proto.messages.*; +import org.eclipse.moquette.spi.impl.subscriptions.Subscription; + +/** + * Default empty implementation that can be used to create custom handlers + * without defining every method. + * + * @author Wagner Macedo + */ +public class DefaultInterceptHandler implements InterceptHandler { + @Override + public void onConnect(ConnectMessage msg) { + } + + @Override + public void onDisconnect(String clientID) { + } + + @Override + public void onPublish(PublishMessage msg) { + } + + @Override + public void onSubscribe(Subscription sub) { + } + + @Override + public void onUnsubscribe(Subscription sub) { + } +} diff --git a/broker/src/main/java/org/eclipse/moquette/interception/InterceptHandler.java b/broker/src/main/java/org/eclipse/moquette/interception/InterceptHandler.java new file mode 100644 index 00000000..97fca7fd --- /dev/null +++ b/broker/src/main/java/org/eclipse/moquette/interception/InterceptHandler.java @@ -0,0 +1,44 @@ +/* + * Copyright (c) 2012-2014 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package org.eclipse.moquette.interception; + +import org.eclipse.moquette.proto.messages.*; +import org.eclipse.moquette.spi.impl.subscriptions.Subscription; + +/** + * This interface is used to inject code for intercepting broker events. + *

+ * The events can act only as observers. + *

+ * Almost every method receives a subclass of {@link AbstractMessage}, except + * onDisconnect that receives the client id string and + * onSubscribe and onUnsubscribe that receive a + * {@link Subscription} object. + * + * @author Wagner Macedo + */ +public interface InterceptHandler { + + void onConnect(ConnectMessage msg); + + void onDisconnect(String clientID); + + void onPublish(PublishMessage msg); + + void onSubscribe(Subscription sub); + + void onUnsubscribe(Subscription sub); +} diff --git a/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java b/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java new file mode 100644 index 00000000..62dfe1f8 --- /dev/null +++ b/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java @@ -0,0 +1,52 @@ +/* + * Copyright (c) 2012-2014 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package org.eclipse.moquette.interception; + +import org.eclipse.moquette.proto.messages.ConnectMessage; +import org.eclipse.moquette.proto.messages.PublishMessage; +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. + *

+ * Interceptor implementations forward notifications to a InterceptHandler, + * that is normally a field. So, the implementations should act as a proxy to a custom + * intercept handler. + * + * @see InterceptHandler + * @author Wagner Macedo + */ +public interface Interceptor { + /** + * When no custom handler is provided, this interceptor should be used. + */ + Interceptor NO_HANDLER_INTERCEPTOR = new NoHandlerInterceptor(); + + void notifyClientConnected(ConnectMessage msg); + + void notifyClientDisconnected(String clientID); + + void notifyTopicPublished(PublishMessage msg); + + void notifyTopicSubscribed(Subscription sub); + + void notifyTopicUnsubscribed(Subscription sub); + +} diff --git a/broker/src/main/java/org/eclipse/moquette/interception/NoHandlerInterceptor.java b/broker/src/main/java/org/eclipse/moquette/interception/NoHandlerInterceptor.java new file mode 100644 index 00000000..b1dcf293 --- /dev/null +++ b/broker/src/main/java/org/eclipse/moquette/interception/NoHandlerInterceptor.java @@ -0,0 +1,33 @@ +package org.eclipse.moquette.interception; + +import org.eclipse.moquette.proto.messages.ConnectMessage; +import org.eclipse.moquette.proto.messages.PublishMessage; +import org.eclipse.moquette.spi.impl.subscriptions.Subscription; + +/** + * @author Wagner Macedo + */ +public class NoHandlerInterceptor implements Interceptor { + NoHandlerInterceptor() { + } + + @Override + public void notifyClientConnected(ConnectMessage msg) { + } + + @Override + public void notifyClientDisconnected(String clientID) { + } + + @Override + public void notifyTopicPublished(PublishMessage msg) { + } + + @Override + public void notifyTopicSubscribed(Subscription sub) { + } + + @Override + public void notifyTopicUnsubscribed(Subscription sub) { + } +} diff --git a/broker/src/main/java/org/eclipse/moquette/server/Server.java b/broker/src/main/java/org/eclipse/moquette/server/Server.java index 5f5a2573..b07ecc9f 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/Server.java +++ b/broker/src/main/java/org/eclipse/moquette/server/Server.java @@ -15,6 +15,7 @@ */ package org.eclipse.moquette.server; +import org.eclipse.moquette.interception.InterceptHandler; import org.eclipse.moquette.server.netty.NettyAcceptor; import org.eclipse.moquette.spi.impl.SimpleMessaging; import org.slf4j.Logger; @@ -86,8 +87,12 @@ public class Server { * */ public void startServer(Properties configProps) throws IOException { - ConfigurationParser confParser = new ConfigurationParser(configProps); + ConfigurationParser confParser = new ConfigurationParser(configProps); m_properties = confParser.getProperties(); + final String handlerProp = System.getProperty("intercept.handler"); + if (handlerProp != null) { + m_properties.setProperty("intercept.handler", handlerProp); + } LOG.info("Persistent store file: " + m_properties.get(PERSISTENT_STORE_PROPERTY_NAME)); messaging = SimpleMessaging.getInstance(); messaging.init(m_properties); 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 new file mode 100644 index 00000000..b39ac197 --- /dev/null +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java @@ -0,0 +1,60 @@ +/* + * Copyright (c) 2012-2014 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package org.eclipse.moquette.spi.impl; + +import org.eclipse.moquette.interception.InterceptHandler; +import org.eclipse.moquette.interception.Interceptor; +import org.eclipse.moquette.proto.messages.ConnectMessage; +import org.eclipse.moquette.proto.messages.PublishMessage; +import org.eclipse.moquette.spi.impl.subscriptions.Subscription; + +/** + * An interceptor that execute the interception tasks asynchronously. + * + * @author Wagner Macedo + */ +final class BrokerInterceptor implements Interceptor { + private final InterceptHandler handler; + + BrokerInterceptor(InterceptHandler handler) { + this.handler = handler; + } + + @Override + public void notifyClientConnected(final ConnectMessage msg) { + handler.onConnect(msg); + } + + @Override + public void notifyClientDisconnected(final String clientID) { + handler.onDisconnect(clientID); + } + + @Override + public void notifyTopicPublished(final PublishMessage msg) { + handler.onPublish(msg); + } + + @Override + public void notifyTopicSubscribed(final Subscription sub) { + handler.onSubscribe(sub); + } + + @Override + public void notifyTopicUnsubscribed(final Subscription sub) { + 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 6cc139ae..5e363c8e 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 @@ -25,6 +25,7 @@ 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; @@ -105,6 +106,8 @@ class ProtocolProcessor implements EventHandler { private IMessagesStore m_messagesStore; private ISessionsStore m_sessionsStore; private IAuthenticator m_authenticator; + private Interceptor interceptor = Interceptor.NO_HANDLER_INTERCEPTOR; + //maps clientID to Will testament, if specified on CONNECT private Map m_willStore = new HashMap<>(); @@ -113,6 +116,13 @@ 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. @@ -122,12 +132,16 @@ class ProtocolProcessor implements EventHandler { * @param authenticator the authenticator used in connect messages. * @param allowAnonymous true connection to clients without credentials. * @param authorizator used to apply ACL policies to publishes and subscriptions. + * @param interceptor to notify events to an intercept handler */ void init(SubscriptionsStore subscriptions, IMessagesStore storageService, ISessionsStore sessionsStore, IAuthenticator authenticator, - boolean allowAnonymous, IAuthorizator authorizator) { + boolean allowAnonymous, IAuthorizator authorizator, Interceptor interceptor) { //m_clientIDs = clientIDs; + if (interceptor != null) { + this.interceptor = interceptor; + } this.subscriptions = subscriptions; this.allowAnonymous = allowAnonymous; m_authorizator = authorizator; @@ -163,6 +177,7 @@ class ProtocolProcessor implements EventHandler { ConnAckMessage okResp = new ConnAckMessage(); okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED); session.write(okResp); + interceptor.notifyClientConnected(msg); return; } @@ -238,6 +253,7 @@ class ProtocolProcessor implements EventHandler { okResp.setSessionPresent(true); } session.write(okResp); + interceptor.notifyClientConnected(msg); LOG.info("Create persistent session for clientID <{}>", msg.getClientID()); m_sessionsStore.addNewSubscription(Subscription.createEmptySubscription(msg.getClientID(), true)); //null means EmptySubscription @@ -301,6 +317,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); } else { LOG.debug("topic {} doesn't have write credentials", topic); } @@ -559,6 +576,7 @@ class ProtocolProcessor implements EventHandler { m_willStore.remove(clientID); LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession); + interceptor.notifyClientDisconnected(clientID); } void processConnectionLost(LostConnectionEvent evt) { 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 46f90b06..e799527d 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 @@ -19,6 +19,8 @@ import com.lmax.disruptor.EventHandler; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.dsl.Disruptor; 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; @@ -55,7 +57,7 @@ import static org.eclipse.moquette.commons.Constants.ACL_FILE_PROPERTY_NAME; * Singleton class that orchestrate the execution of the protocol. * * Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion; - * the requests income from front Netty connectors and are dispatched to the + * the requests income from front Netty connectors and are dispatched to the * ProtocolProcessor. * * @author andrea @@ -63,9 +65,9 @@ import static org.eclipse.moquette.commons.Constants.ACL_FILE_PROPERTY_NAME; public class SimpleMessaging implements IMessaging, EventHandler { private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class); - + private SubscriptionsStore subscriptions; - + private RingBuffer m_ringBuffer; private IMessagesStore m_storageService; @@ -73,6 +75,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { private ExecutorService m_executor; private Disruptor m_disruptor; + private Interceptor interceptor; private static SimpleMessaging INSTANCE; @@ -192,9 +195,16 @@ public class SimpleMessaging implements IMessaging, EventHandler { m_storageService.initStore(); + try { + final InterceptHandler handler = Class.forName(props.getProperty("intercept.handler")) + .asSubclass(InterceptHandler.class).newInstance(); + interceptor = new BrokerInterceptor(handler); + } catch (ClassNotFoundException | InstantiationException | IllegalAccessException | NullPointerException | ClassCastException ignored) { + } + //List storedSubscriptions = m_sessionsStore.listAllSubscriptions(); //subscriptions.init(storedSubscriptions); - subscriptions.init(m_sessionsStore); + subscriptions.init(m_sessionsStore, interceptor); String configPath = System.getProperty("moquette.path", null); String authenticatorClassName = props.getProperty(Constants.AUTHENTICATOR_CLASS_NAME, ""); @@ -240,7 +250,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); + m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator, allowAnonymous, authorizator, interceptor); } private Object loadClass(String className, Class cls) { 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 dff1c5ea..8b1a5e53 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,6 +19,7 @@ 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; @@ -30,6 +31,8 @@ import org.slf4j.LoggerFactory; */ public class SubscriptionsStore { + private Interceptor interceptor = Interceptor.NO_HANDLER_INTERCEPTOR; + /** * Check if the topic filter of the subscription is well formed * */ @@ -96,9 +99,20 @@ public class SubscriptionsStore { /** * Initialize the subscription tree with the list of subscriptions. + * Maintained for compatibility reasons. */ public void init(ISessionsStore sessionsStore) { + init(sessionsStore, null); + } + + /** + * Initialize the subscription tree with the list of subscriptions and an interceptor. + */ + public void init(ISessionsStore sessionsStore, Interceptor interceptor) { LOG.debug("init invoked"); + if (interceptor != null) { + this.interceptor = interceptor; + } m_sessionsStore = sessionsStore; List subscriptions = sessionsStore.listAllSubscriptions(); //reload any subscriptions persisted @@ -118,6 +132,7 @@ public class SubscriptionsStore { protected void addDirect(Subscription newSubscription) { TreeNode current = findMatchingNode(newSubscription.topicFilter); current.addSubscription(newSubscription); + interceptor.notifyTopicSubscribed(newSubscription); } private TreeNode findMatchingNode(String topic) { @@ -171,6 +186,7 @@ public class SubscriptionsStore { if (toBeRemoved != null) { matchNode.subscriptions().remove(toBeRemoved); + interceptor.notifyTopicUnsubscribed(toBeRemoved); } } @@ -184,6 +200,7 @@ public class SubscriptionsStore { List allSubscriptions = subsCollector.getResult(); for (Subscription subscription : allSubscriptions) { removeSubscription(subscription.getTopicFilter(), subscription.getClientId()); + interceptor.notifyTopicUnsubscribed(subscription); } } @@ -191,9 +208,16 @@ public class SubscriptionsStore { * Visit the topics tree to remove matching subscriptions with clientID */ public void removeForClient(String clientID) { - subscriptions.removeClientSubscriptions(clientID); + final List clientSubscriptions = new ArrayList<>(); + + subscriptions.removeClientSubscriptions(clientID, clientSubscriptions); //persist the update m_sessionsStore.wipeSubscriptions(clientID); + + // Notify the unsubscribe event + for (Subscription sub : clientSubscriptions) { + interceptor.notifyTopicUnsubscribed(sub); + } } public void deactivate(String clientID) { diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java index 2a165ff0..70c6a738 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/TreeNode.java @@ -130,7 +130,7 @@ class TreeNode { return res; } - void removeClientSubscriptions(String clientID) { + void removeClientSubscriptions(String clientID, List collect) { //collect what to delete and then delete to avoid ConcurrentModification List subsToRemove = new ArrayList<>(); for (Subscription s : m_subscriptions) { @@ -143,9 +143,11 @@ class TreeNode { m_subscriptions.remove(s); } + collect.addAll(subsToRemove); + //go deep for (TreeNode child : m_children) { - child.removeClientSubscriptions(clientID); + child.removeClientSubscriptions(clientID, collect); } }