From 03c0097641ddc8adb6c541f28a7a1cbdc3d50fc0 Mon Sep 17 00:00:00 2001 From: bcw Date: Fri, 21 Oct 2016 10:28:36 +0800 Subject: [PATCH 1/2] broker: notify when client connection lost --- .../interception/InterceptHandler.java | 9 +++----- .../io/moquette/interception/Interceptor.java | 2 ++ .../InterceptConnectionLostMessage.java | 22 +++++++++++++++++++ .../moquette/spi/impl/BrokerInterceptor.java | 12 ++++++++++ .../moquette/spi/impl/ProtocolProcessor.java | 3 +++ 5 files changed, 42 insertions(+), 6 deletions(-) create mode 100644 broker/src/main/java/io/moquette/interception/messages/InterceptConnectionLostMessage.java 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); } /** From 3e89bc5d2ccbddd1b354c0dad95a2e1a80c5b3dc Mon Sep 17 00:00:00 2001 From: bcw Date: Mon, 24 Oct 2016 11:37:10 +0800 Subject: [PATCH 2/2] =?UTF-8?q?notify=20when=20client=20connection=20lost?= =?UTF-8?q?=EF=BC=9B=20implement=20onConnectionLost()=20method=20in=20Abst?= =?UTF-8?q?ractInterceptHandler.java;?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../io/moquette/interception/AbstractInterceptHandler.java | 3 +++ .../java/io/moquette/spi/impl/BrokerInterceptorTest.java | 5 +++++ 2 files changed, 8 insertions(+) diff --git a/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java b/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java index b64c03e6..c1a75447 100644 --- a/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java +++ b/broker/src/main/java/io/moquette/interception/AbstractInterceptHandler.java @@ -30,6 +30,9 @@ public abstract class AbstractInterceptHandler implements InterceptHandler { @Override public void onDisconnect(InterceptDisconnectMessage msg) {} + @Override + public void onConnectionLost(InterceptConnectionLostMessage msg) {} + @Override public void onPublish(InterceptPublishMessage msg) {} diff --git a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java index b84b4896..3a5fec48 100644 --- a/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java +++ b/broker/src/test/java/io/moquette/spi/impl/BrokerInterceptorTest.java @@ -51,6 +51,11 @@ public class BrokerInterceptorTest { n.set(50); } + @Override + public void onConnectionLost(InterceptConnectionLostMessage msg) { + n.set(30); + } + @Override public void onPublish(InterceptPublishMessage msg) { n.set(60);