diff --git a/broker/src/main/java/io/moquette/interception/InterceptHandler.java b/broker/src/main/java/io/moquette/interception/InterceptHandler.java index c8e905e6..2a3fb909 100644 --- a/broker/src/main/java/io/moquette/interception/InterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/InterceptHandler.java @@ -15,14 +15,9 @@ */ package io.moquette.interception; +import io.moquette.interception.messages.*; import io.moquette.parser.proto.messages.AbstractMessage; import io.moquette.spi.impl.subscriptions.Subscription; -import io.moquette.interception.messages.InterceptConnectMessage; -import io.moquette.interception.messages.InterceptAcknowledgedMessage; -import io.moquette.interception.messages.InterceptDisconnectMessage; -import io.moquette.interception.messages.InterceptPublishMessage; -import io.moquette.interception.messages.InterceptSubscribeMessage; -import io.moquette.interception.messages.InterceptUnsubscribeMessage; /** * This interface is used to inject code for intercepting broker events. @@ -42,6 +37,8 @@ public interface InterceptHandler { void onDisconnect(InterceptDisconnectMessage msg); + void onConnectionLost(InterceptConnectionLostMessage msg); + void onPublish(InterceptPublishMessage msg); void onSubscribe(InterceptSubscribeMessage msg); diff --git a/broker/src/main/java/io/moquette/interception/Interceptor.java b/broker/src/main/java/io/moquette/interception/Interceptor.java index 634b1748..71bb8b3d 100644 --- a/broker/src/main/java/io/moquette/interception/Interceptor.java +++ b/broker/src/main/java/io/moquette/interception/Interceptor.java @@ -39,6 +39,8 @@ public interface Interceptor { void notifyClientDisconnected(String clientID, String username); + void notifyClientConnectionLost(String clientID, String username); + void notifyTopicPublished(PublishMessage msg, String clientID, final String username); void notifyTopicSubscribed(Subscription sub, final String username); diff --git a/broker/src/main/java/io/moquette/interception/messages/InterceptConnectionLostMessage.java b/broker/src/main/java/io/moquette/interception/messages/InterceptConnectionLostMessage.java new file mode 100644 index 00000000..0c4b662b --- /dev/null +++ b/broker/src/main/java/io/moquette/interception/messages/InterceptConnectionLostMessage.java @@ -0,0 +1,22 @@ +package io.moquette.interception.messages; + +/** + * @author Wagner Macedo + */ +public class InterceptConnectionLostMessage { + private final String clientID; + private final String username; + + public InterceptConnectionLostMessage(String clientID, String username) { + this.clientID = clientID; + this.username = username; + } + + public String getClientID() { + return clientID; + } + + public String getUsername() { + return username; + } +} 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 81899281..481bb547 100644 --- a/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/io/moquette/spi/impl/BrokerInterceptor.java @@ -72,6 +72,18 @@ final class BrokerInterceptor implements Interceptor { } } + @Override + public void notifyClientConnectionLost(final String clientID, final String username) { + for (final InterceptHandler handler : this.handlers) { + executor.execute(new Runnable() { + @Override + public void run() { + handler.onConnectionLost(new InterceptConnectionLostMessage(clientID, username)); + } + }); + } + } + @Override public void notifyTopicPublished(final PublishMessage msg, final String clientID, final String username) { for (final InterceptHandler handler : this.handlers) { diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 5199b19c..c5e89f58 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java @@ -704,6 +704,9 @@ public class ProtocolProcessor { forwardPublishWill(will, clientID); m_willStore.remove(clientID); } + + String username = NettyUtils.userName(channel); + m_interceptor.notifyClientConnectionLost(clientID, username); } /**