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 02bb2ac5..44b0013e 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,9 @@ 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.concurrent.ExecutorService; +import java.util.concurrent.Executors; + /** * An interceptor that execute the interception tasks asynchronously. * @@ -28,33 +31,67 @@ import org.eclipse.moquette.spi.impl.subscriptions.Subscription; */ final class BrokerInterceptor implements Interceptor { private final InterceptHandler handler; + private final ExecutorService executor; BrokerInterceptor(InterceptHandler handler) { this.handler = handler; + executor = Executors.newFixedThreadPool(1); + } + + /** + * Shutdown graciously the executor service + */ + void stop() { + executor.shutdown(); } @Override public void notifyClientConnected(final ConnectMessage msg) { - handler.onConnect(msg.readOnlyClone()); + executor.execute(new Runnable() { + @Override + public void run() { + handler.onConnect(msg.readOnlyClone()); + } + }); } @Override public void notifyClientDisconnected(final String clientID) { - handler.onDisconnect(clientID); + executor.execute(new Runnable() { + @Override + public void run() { + handler.onDisconnect(clientID); + } + }); } @Override public void notifyTopicPublished(final PublishMessage msg) { - handler.onPublish(msg.readOnlyClone()); + executor.execute(new Runnable() { + @Override + public void run() { + handler.onPublish(msg.readOnlyClone()); + } + }); } @Override public void notifyTopicSubscribed(final Subscription sub) { - handler.onSubscribe(sub.clone()); + executor.execute(new Runnable() { + @Override + public void run() { + handler.onSubscribe(sub.clone()); + } + }); } @Override public void notifyTopicUnsubscribed(final Subscription sub) { - handler.onUnsubscribe(sub.clone()); + executor.execute(new Runnable() { + @Override + public void run() { + handler.onUnsubscribe(sub.clone()); + } + }); } } 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 e799527d..0eb81d18 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 @@ -290,6 +290,9 @@ public class SimpleMessaging implements IMessaging, EventHandler { private void processStop() { LOG.debug("processStop invoked"); + if (interceptor != null) { + ((BrokerInterceptor) interceptor).stop(); + } m_storageService.close(); LOG.debug("subscription tree {}", subscriptions.dumpTree()); // m_eventProcessor.halt();