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 bab6378e..7d2c1086 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -13,6 +13,7 @@ * * You may elect to redistribute this code under either of these licenses. */ + package io.moquette.spi.impl; import io.moquette.BrokerConstants; @@ -25,7 +26,6 @@ import io.netty.handler.codec.mqtt.MqttConnectMessage; import io.netty.handler.codec.mqtt.MqttPublishMessage; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import java.util.HashMap; import java.util.List; import java.util.Map; @@ -33,7 +33,6 @@ import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; - import static io.moquette.logging.LoggingUtils.getInterceptorIds; /** @@ -47,56 +46,60 @@ final class BrokerInterceptor implements Interceptor { 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()); - } - for (InterceptHandler handler : handlers) { - this.addInterceptHandler(handler); - } - executor = Executors.newFixedThreadPool(poolSize); - } - - /** - * Configures a broker interceptor, with a thread pool of one thread. - * @param handlers - */ - public BrokerInterceptor(List handlers) { - this(1, handlers); - } - - /** - * Configures a broker interceptor using the pool size specified in the - * IConfig argument. - */ - public BrokerInterceptor(IConfig props, List handlers) { - this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers); - } + 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()); + } + for (InterceptHandler handler : handlers) { + this.addInterceptHandler(handler); + } + executor = Executors.newFixedThreadPool(poolSize); + } + + /** + * Configures a broker interceptor, with a thread pool of one thread. + * + * @param handlers + */ + public BrokerInterceptor(List handlers) { + this(1, handlers); + } + + /** + * Configures a broker interceptor using the pool size specified in the IConfig argument. + */ + public BrokerInterceptor(IConfig props, List handlers) { + this(Integer.parseInt(props.getProperty(BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE, "1")), handlers); + } /** * 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(); - } - } + 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(); + } + } @Override public void notifyClientConnected(final MqttConnectMessage msg) { for (final InterceptHandler handler : this.handlers.get(InterceptConnectMessage.class)) { - LOG.debug("Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.", - msg.payload().clientIdentifier(), handler.getID()); + LOG.debug( + "Sending MQTT CONNECT message to interceptor. MqttClientId = {}, interceptorId = {}.", + msg.payload().clientIdentifier(), + handler.getID()); executor.execute(new Runnable() { + @Override public void run() { handler.onConnect(new InterceptConnectMessage(msg)); @@ -106,11 +109,16 @@ final class BrokerInterceptor implements Interceptor { } @Override - public void notifyClientDisconnected(final String clientID, final String username ) { + public void notifyClientDisconnected(final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptDisconnectMessage.class)) { - LOG.debug("Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, " + - "interceptorId = {}.", clientID, username, handler.getID()); + LOG.debug( + "Notifying MQTT client disconnection to interceptor. MqttClientId = {}, username = {}, " + + "interceptorId = {}.", + clientID, + username, + handler.getID()); executor.execute(new Runnable() { + @Override public void run() { handler.onDisconnect(new InterceptDisconnectMessage(clientID, username)); @@ -122,9 +130,14 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyClientConnectionLost(final String clientID, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptConnectionLostMessage.class)) { - LOG.debug("Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, " + - "username = {}, interceptorId = {}.", clientID, username, handler.getID()); + LOG.debug( + "Notifying unexpected MQTT client disconnection to interceptor. MqttClientId = {}, " + + "username = {}, interceptorId = {}.", + clientID, + username, + handler.getID()); executor.execute(new Runnable() { + @Override public void run() { handler.onConnectionLost(new InterceptConnectionLostMessage(clientID, username)); @@ -138,9 +151,15 @@ final class BrokerInterceptor implements Interceptor { int messageId = msg.variableHeader().messageId(); String topic = msg.variableHeader().topicName(); for (final InterceptHandler handler : this.handlers.get(InterceptPublishMessage.class)) { - LOG.debug("Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " + - "interceptorId = {}.", clientID, messageId, topic, handler.getID()); + LOG.debug( + "Notifying MQTT PUBLISH message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " + + "interceptorId = {}.", + clientID, + messageId, + topic, + handler.getID()); executor.execute(new Runnable() { + @Override public void run() { handler.onPublish(new InterceptPublishMessage(msg, clientID, username)); @@ -152,9 +171,14 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyTopicSubscribed(final Subscription sub, final String username) { for (final InterceptHandler handler : this.handlers.get(InterceptSubscribeMessage.class)) { - LOG.debug("Notifying MQTT SUBSCRIBE message to interceptor. MqttClientId = {}, topicFilter = {}, " + - "interceptorId = {}.", sub.getClientId(), sub.getTopicFilter(), handler.getID()); + 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() { handler.onSubscribe(new InterceptSubscribeMessage(sub, username)); @@ -166,9 +190,14 @@ 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)) { - LOG.debug("Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, " + - "interceptorId = {}.", clientID, topic, handler.getID()); + LOG.debug( + "Notifying MQTT UNSUBSCRIBE message to interceptor. MqttClientId = {}, topic = {}, " + + "interceptorId = {}.", + clientID, + topic, + handler.getID()); executor.execute(new Runnable() { + @Override public void run() { handler.onUnsubscribe(new InterceptUnsubscribeMessage(topic, clientID, username)); @@ -177,13 +206,18 @@ final class BrokerInterceptor implements Interceptor { } } - @Override - public void notifyMessageAcknowledged( final InterceptAcknowledgedMessage msg ) { + @Override + public void notifyMessageAcknowledged(final InterceptAcknowledgedMessage msg) { for (final InterceptHandler handler : this.handlers.get(InterceptAcknowledgedMessage.class)) { - LOG.debug("Notifying MQTT ACK message to interceptor. MqttClientId = {}, messageId = {}, topic = {}, " + - "interceptorId = {}.", - msg.getMsg().getClientID(), msg.getMsg().getMessageID(), msg.getTopic(), handler.getID()); + 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() { handler.onMessageAcknowledged(msg); @@ -192,31 +226,35 @@ final class BrokerInterceptor implements Interceptor { } } - @Override - public void addInterceptHandler(InterceptHandler 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) { - 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); - } + @Override + public void addInterceptHandler(InterceptHandler 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); + } } - private static Class[] getInterceptedMessageTypes(InterceptHandler interceptHandler) { - Class interceptedMessageTypes[] = interceptHandler.getInterceptedMessageTypes(); - if (interceptedMessageTypes == null) { - return InterceptHandler.ALL_MESSAGE_TYPES; - } - return interceptedMessageTypes; - } + @Override + public void removeInterceptHandler(InterceptHandler 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); + } + } + + private static Class[] getInterceptedMessageTypes(InterceptHandler interceptHandler) { + Class[] interceptedMessageTypes = interceptHandler.getInterceptedMessageTypes(); + if (interceptedMessageTypes == null) { + return InterceptHandler.ALL_MESSAGE_TYPES; + } + return interceptedMessageTypes; + } }