diff --git a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java index a523cf86..5a49fae3 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -38,16 +38,23 @@ import io.moquette.parser.proto.messages.PublishMessage; import io.moquette.server.config.IConfig; import io.moquette.spi.impl.subscriptions.Subscription; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import static io.moquette.logging.LoggingUtils.getInterceptorIds; + /** * An interceptor that execute the interception tasks asynchronously. * * @author Wagner Macedo */ final class BrokerInterceptor implements Interceptor { + + private static final Logger LOG = LoggerFactory.getLogger(BrokerInterceptor.class); private final Map, List> handlers; private final ExecutorService executor; private BrokerInterceptor(int poolSize, List handlers) { + LOG.info("Initializing broker interceptor. InterceptorIds = {}.", getInterceptorIds(handlers)); this.handlers = new HashMap<>(); for (Class messageType : InterceptHandler.ALL_MESSAGE_TYPES) { this.handlers.put(messageType, new CopyOnWriteArrayList()); @@ -78,11 +85,14 @@ final class BrokerInterceptor implements Interceptor { * Shutdown graciously the executor service */ void stop() { + LOG.info("Shutting down interceptor thread pool..."); executor.shutdown(); try { + LOG.info("Waiting for thread pool tasks to terminate..."); executor.awaitTermination(10L, TimeUnit.SECONDS); } catch (InterruptedException e) {} if (!executor.isTerminated()) { + LOG.warn("Forcing shutdown of interceptor thread pool..."); executor.shutdownNow(); } } @@ -90,6 +100,10 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientConnected(final ConnectMessage msg) { for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) { + if (LOG.isDebugEnabled()) { + LOG.debug("Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.", + msg.getClientID(), handler.getID()); + } executor.execute(new Runnable() { @Override public void run() { @@ -102,6 +116,10 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientDisconnected(final String clientID, final String username ) { for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) { + if (LOG.isDebugEnabled()) { + LOG.debug("Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, interceptorId = {}.", + clientID, username, handler.getID()); + } executor.execute(new Runnable() { @Override public void run() { @@ -114,6 +132,10 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientConnectionLost(final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) { + if (LOG.isDebugEnabled()) { + LOG.debug("Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, interceptorId = {}.", + clientID, username, handler.getID()); + } executor.execute(new Runnable() { @Override public void run() { @@ -126,6 +148,10 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyTopicPublished(final PublishMessage msg, final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) { + if (LOG.isDebugEnabled()) { + LOG.debug("Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, interceptorId = {}.", + clientID, msg.getMessageID(), msg.getTopicName(), handler.getID()); + } executor.execute(new Runnable() { @Override public void run() { @@ -138,6 +164,10 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyTopicSubscribed(final Subscription sub, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) { + if (LOG.isDebugEnabled()) { + LOG.debug("Notifying MQTT SUBSCRIBE message to interceptor. MqttClientId = {}, topicFilter = {}, interceptorId = {}.", + sub.getClientId(), sub.getTopicFilter(), handler.getID()); + } executor.execute(new Runnable() { @Override public void run() { @@ -150,6 +180,10 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyTopicUnsubscribed(final String topic, final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptUnsubscribeMessage.class)) { + if (LOG.isDebugEnabled()) { + LOG.debug("Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, interceptorId = {}.", + clientID, topic, handler.getID()); + } executor.execute(new Runnable() { @Override public void run() { @@ -162,6 +196,10 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyMessageAcknowledged( final InterceptAcknowledgedMessage msg ) { for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) { + if (LOG.isDebugEnabled()) { + LOG.debug("Notifying MQTT ACK message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, interceptorId = {}.", + msg.getMsg().getClientID(), msg.getMsg().getMessageID(), msg.getTopic(), handler.getID()); + } executor.execute(new Runnable() { @Override public void run() { @@ -173,14 +211,20 @@ final class BrokerInterceptor implements Interceptor { @Override public void addInterceptHandler(InterceptHandler interceptHandler) { - for (Class interceptMessageType : getInterceptedMessageTypes(interceptHandler)) { + Class[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler); + LOG.info("Adding MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.", + interceptHandler.getID(), interceptedMessageTypes); + for (Class interceptMessageType : interceptedMessageTypes) { this.handlers.get(interceptMessageType).add(interceptHandler); } } @Override public void removeInterceptHandler(InterceptHandler interceptHandler) { - for (Class interceptMessageType : getInterceptedMessageTypes(interceptHandler)) { + Class[] interceptedMessageTypes = getInterceptedMessageTypes(interceptHandler); + LOG.info("Removing MQTT message interceptor. InterceptorId = {}, handledMessageTypes = {}.", + interceptHandler.getID(), interceptedMessageTypes); + for (Class interceptMessageType : interceptedMessageTypes) { this.handlers.get(interceptMessageType).remove(interceptHandler); } }