Enabling ability for interception of broker events

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) {
+    }
+}

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.
+ * <p>
+ * The events can act only as observers.
+ * <p>
+ * Almost every method receives a subclass of {@link AbstractMessage}, except
+ * <code>onDisconnect</code> that receives the client id string and
+ * <code>onSubscribe</code> and <code>onUnsubscribe</code> 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);
+}

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.
+ * <p>
+ * 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.
+ * <p>
+ * Interceptor implementations forward notifications to a <code>InterceptHandler</code>,
+ * 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);
+
+}

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) {
+    }
+}

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 {
      * </ul>
      */
     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);

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);
+    }
+}

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<ValueEvent> {
     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<String, WillMessage> m_willStore = new HashMap<>();
     
@@ -113,6 +116,13 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
 
     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<ValueEvent> {
      * @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<ValueEvent> {
             ConnAckMessage okResp = new ConnAckMessage();
             okResp.setReturnCode(ConnAckMessage.IDENTIFIER_REJECTED);
             session.write(okResp);
+            interceptor.notifyClientConnected(msg);
             return;
         }
 
@@ -238,6 +253,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
             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<ValueEvent> {
         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<ValueEvent> {
         m_willStore.remove(clientID);
         
         LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession);
+        interceptor.notifyClientDisconnected(clientID);
     }
     
     void processConnectionLost(LostConnectionEvent evt) {

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<ValueEvent> {
 
     private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class);
-    
+
     private SubscriptionsStore subscriptions;
-    
+
     private RingBuffer<ValueEvent> m_ringBuffer;
 
     private IMessagesStore m_storageService;
@@ -73,6 +75,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
 
     private ExecutorService m_executor;
     private Disruptor<ValueEvent> m_disruptor;
+    private Interceptor interceptor;
 
     private static SimpleMessaging INSTANCE;
     
@@ -192,9 +195,16 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
 
         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<Subscription> 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<ValueEvent> {
         }
 
         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) {

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<Subscription> 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<Subscription> 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<Subscription> 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) {

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<Subscription> collect) {
         //collect what to delete and then delete to avoid ConcurrentModification
         List<Subscription> 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);
         }
     }