Merge pull request #224 from bcw104/master

broker: notify when client connection lost

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) {}
 

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);

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);

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;
+	}
+}

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) {

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);
     }
 
     /**

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);