Removed unsufull classes and moved out of internal broker structures the firing of interceptor events

broker/src/main/java/org/eclipse/moquette/interception/DefaultInterceptHandler.java

@@ -1,47 +0,0 @@
-/*
- * 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

@@ -40,5 +40,5 @@ public interface InterceptHandler {
 
     void onSubscribe(Subscription sub);
 
-    void onUnsubscribe(Subscription sub);
+    void onUnsubscribe(String topic);
 }

broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java

@@ -37,7 +37,6 @@ public interface Interceptor {
     /**
      * When no custom handler is provided, this interceptor should be used.
      */
-    Interceptor NO_HANDLER_INTERCEPTOR = new NoHandlerInterceptor();
 
     void notifyClientConnected(ConnectMessage msg);
 
@@ -47,6 +46,6 @@ public interface Interceptor {
 
     void notifyTopicSubscribed(Subscription sub);
 
-    void notifyTopicUnsubscribed(Subscription sub);
+    void notifyTopicUnsubscribed(String topic);
 
 }

broker/src/main/java/org/eclipse/moquette/interception/NoHandlerInterceptor.java

@@ -1,33 +0,0 @@
-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/spi/impl/BrokerInterceptor.java

@@ -86,11 +86,11 @@ final class BrokerInterceptor implements Interceptor {
     }
 
     @Override
-    public void notifyTopicUnsubscribed(final Subscription sub) {
+    public void notifyTopicUnsubscribed(final String sub) {
         executor.execute(new Runnable() {
             @Override
             public void run() {
-                handler.onUnsubscribe(sub.clone());
+                handler.onUnsubscribe(sub);
             }
         });
     }

broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java

@@ -106,7 +106,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     private IMessagesStore m_messagesStore;
     private ISessionsStore m_sessionsStore;
     private IAuthenticator m_authenticator;
-    private Interceptor interceptor = Interceptor.NO_HANDLER_INTERCEPTOR;
+    private Interceptor interceptor = null;
 
     //maps clientID to Will testament, if specified on CONNECT
     private Map<String, WillMessage> m_willStore = new HashMap<>();
@@ -610,6 +610,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         for (String topic : topics) {
             subscriptions.removeSubscription(topic, clientID);
             m_sessionsStore.removeSubscription(topic, clientID);
+            interceptor.notifyTopicUnsubscribed(topic);
         }
 
         //ack the client
@@ -653,6 +654,9 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         m_sessionsStore.addNewSubscription(newSubscription);
         subscriptions.add(newSubscription);
 
+        //notify the Observables
+        interceptor.notifyTopicSubscribed(newSubscription);
+
         //scans retained messages to be published to the new subscription
         Collection<IMessagesStore.StoredMessage> messages = m_messagesStore.searchMatching(new IMatchingCondition() {
             public boolean match(String key) {

broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java

@@ -199,12 +199,13 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
             final InterceptHandler handler = Class.forName(props.getProperty("intercept.handler"))
                     .asSubclass(InterceptHandler.class).newInstance();
             interceptor = new BrokerInterceptor(handler);
-        } catch (ClassNotFoundException | InstantiationException | IllegalAccessException | NullPointerException | ClassCastException ignored) {
+        } catch (Throwable ex) {
+            LOG.error("Can't load the intercept handler {}", ex);
         }
 
         //List<Subscription> storedSubscriptions = m_sessionsStore.listAllSubscriptions();
         //subscriptions.init(storedSubscriptions);
-        subscriptions.init(m_sessionsStore, interceptor);
+        subscriptions.init(m_sessionsStore);
 
         String configPath = System.getProperty("moquette.path", null);
         String authenticatorClassName = props.getProperty(Constants.AUTHENTICATOR_CLASS_NAME, "");

broker/src/main/java/org/eclipse/moquette/spi/impl/subscriptions/SubscriptionsStore.java

@@ -31,8 +31,6 @@ 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
      * */
@@ -102,17 +100,7 @@ public class SubscriptionsStore {
      * 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
@@ -132,7 +120,6 @@ public class SubscriptionsStore {
     protected void addDirect(Subscription newSubscription) {
         TreeNode current = findMatchingNode(newSubscription.topicFilter);
         current.addSubscription(newSubscription);
-        interceptor.notifyTopicSubscribed(newSubscription);
     }
     
     private TreeNode findMatchingNode(String topic) {
@@ -186,12 +173,12 @@ public class SubscriptionsStore {
         
         if (toBeRemoved != null) {
             matchNode.subscriptions().remove(toBeRemoved);
-            interceptor.notifyTopicUnsubscribed(toBeRemoved);
         }
     }
     
     /**
      * TODO implement testing
+     * It's public because needed in tests :-( bleah
      */
     public void clearAllSubscriptions() {
         SubscriptionTreeCollector subsCollector = new SubscriptionTreeCollector();
@@ -200,7 +187,6 @@ public class SubscriptionsStore {
         List<Subscription> allSubscriptions = subsCollector.getResult();
         for (Subscription subscription : allSubscriptions) {
             removeSubscription(subscription.getTopicFilter(), subscription.getClientId());
-            interceptor.notifyTopicUnsubscribed(subscription);
         }
     }
 
@@ -208,16 +194,9 @@ public class SubscriptionsStore {
      * Visit the topics tree to remove matching subscriptions with clientID
      */
     public void removeForClient(String clientID) {
-        final List<Subscription> clientSubscriptions = new ArrayList<>();
+        subscriptions.removeClientSubscriptions(clientID);
-
-        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, List<Subscription> collect) {
+    void removeClientSubscriptions(String clientID) {
         //collect what to delete and then delete to avoid ConcurrentModification
         List<Subscription> subsToRemove = new ArrayList<>();
         for (Subscription s : m_subscriptions) {
@@ -143,11 +143,9 @@ class TreeNode {
             m_subscriptions.remove(s);
         }
 
-        collect.addAll(subsToRemove);
-
         //go deep
         for (TreeNode child : m_children) {
-            child.removeClientSubscriptions(clientID, collect);
+            child.removeClientSubscriptions(clientID);
         }
     }
 

broker/src/test/java/org/eclipse/moquette/spi/impl/BrokerInterceptorTest.java

@@ -58,7 +58,7 @@ public class BrokerInterceptorTest {
         }
 
         @Override
-        public void onUnsubscribe(Subscription sub) {
+        public void onUnsubscribe(String topic) {
             n.set(80);
         }
     });
@@ -109,7 +109,7 @@ public class BrokerInterceptorTest {
 
     @Test
     public void testNotifyTopicUnsubscribed() throws Exception {
-        interceptor.notifyTopicUnsubscribed(new Subscription("cli1", "o2", AbstractMessage.QOSType.MOST_ONE, true));
+        interceptor.notifyTopicUnsubscribed("o2");
         interval();
         assertEquals(80, n.get());
     }