From 9daff2b7381f92e535003f1dfbc7d0f911afd8e5 Mon Sep 17 00:00:00 2001 From: andrea Date: Sat, 8 Aug 2015 11:29:58 +0200 Subject: [PATCH] Removed unsufull classes and moved out of internal broker structures the firing of interceptor events --- .../interception/DefaultInterceptHandler.java | 47 ------------------- .../interception/InterceptHandler.java | 2 +- .../moquette/interception/Interceptor.java | 3 +- .../interception/NoHandlerInterceptor.java | 33 ------------- .../moquette/spi/impl/BrokerInterceptor.java | 4 +- .../moquette/spi/impl/ProtocolProcessor.java | 6 ++- .../moquette/spi/impl/SimpleMessaging.java | 5 +- .../subscriptions/SubscriptionsStore.java | 25 +--------- .../spi/impl/subscriptions/TreeNode.java | 6 +-- .../spi/impl/BrokerInterceptorTest.java | 4 +- 10 files changed, 18 insertions(+), 117 deletions(-) delete mode 100644 broker/src/main/java/org/eclipse/moquette/interception/DefaultInterceptHandler.java delete mode 100644 broker/src/main/java/org/eclipse/moquette/interception/NoHandlerInterceptor.java diff --git a/broker/src/main/java/org/eclipse/moquette/interception/DefaultInterceptHandler.java b/broker/src/main/java/org/eclipse/moquette/interception/DefaultInterceptHandler.java deleted file mode 100644 index 44cbd748..00000000 --- a/broker/src/main/java/org/eclipse/moquette/interception/DefaultInterceptHandler.java +++ /dev/null @@ -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) { - } -} diff --git a/broker/src/main/java/org/eclipse/moquette/interception/InterceptHandler.java b/broker/src/main/java/org/eclipse/moquette/interception/InterceptHandler.java index 97fca7fd..2be6f472 100644 --- a/broker/src/main/java/org/eclipse/moquette/interception/InterceptHandler.java +++ b/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); } 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 62dfe1f8..26c0021c 100644 --- a/broker/src/main/java/org/eclipse/moquette/interception/Interceptor.java +++ b/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); } diff --git a/broker/src/main/java/org/eclipse/moquette/interception/NoHandlerInterceptor.java b/broker/src/main/java/org/eclipse/moquette/interception/NoHandlerInterceptor.java deleted file mode 100644 index b1dcf293..00000000 --- a/broker/src/main/java/org/eclipse/moquette/interception/NoHandlerInterceptor.java +++ /dev/null @@ -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) { - } -} 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 44b0013e..de393e87 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 @@ -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); } }); } 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 5e363c8e..3dc73f29 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 @@ -106,7 +106,7 @@ class ProtocolProcessor implements EventHandler { 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 m_willStore = new HashMap<>(); @@ -610,6 +610,7 @@ class ProtocolProcessor implements EventHandler { 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 { m_sessionsStore.addNewSubscription(newSubscription); subscriptions.add(newSubscription); + //notify the Observables + interceptor.notifyTopicSubscribed(newSubscription); + //scans retained messages to be published to the new subscription Collection messages = m_messagesStore.searchMatching(new IMatchingCondition() { public boolean match(String key) { 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 0eb81d18..8ad92961 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 @@ -199,12 +199,13 @@ public class SimpleMessaging implements IMessaging, EventHandler { 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 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, ""); 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 8b1a5e53..ab210678 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 @@ -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 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 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 clientSubscriptions = new ArrayList<>(); - - subscriptions.removeClientSubscriptions(clientID, clientSubscriptions); + subscriptions.removeClientSubscriptions(clientID); //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 70c6a738..2a165ff0 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, List collect) { + void removeClientSubscriptions(String clientID) { //collect what to delete and then delete to avoid ConcurrentModification List 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); } } 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 2a6bc70c..11c3a712 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 @@ -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()); }