Added handling of dirty connection close from client, without an DISCONNECT message

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/IMessaging.java

@@ -13,6 +13,8 @@ public interface IMessaging {
     void stop();
 
     void disconnect(ServerChannel session);
+    
+    void lostConnection(String clientID);
 
     void handleProtocolMessage(ServerChannel session, AbstractMessage msg);
 }

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/ProtocolProcessor.java

@@ -192,6 +192,9 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         final AbstractMessage.QOSType qos = evt.getQos();
         final byte[] message = evt.getMessage();
         boolean retain = evt.isRetain();
+        
+        LOG.info(String.format("Publish recieved from clientID <%s> on topic <%s> with QoS %s, content [%s]", 
+                evt.getClientID(), evt.getTopic(), evt.getQos(), new String(evt.getMessage())));
 
         String publishKey = null;
         if (qos == AbstractMessage.QOSType.LEAST_ONE) {
@@ -227,7 +230,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     private void publish2Subscribers(String topic, AbstractMessage.QOSType qos, byte[] message, boolean retain, Integer messageID) {
         LOG.debug("publish2Subscribers republishing to existing subscribers that matches the topic " + topic);
         for (final Subscription sub : subscriptions.matches(topic)) {
-            LOG.debug("found matching subscription on topic " + sub.getTopic());
+            LOG.debug(String.format("Broker publishing to client <%s> topic <%s> qos <%s> content [%s]", 
+                    sub.getClientId(), sub.getTopic(), qos, new String(message)));
             if (qos == AbstractMessage.QOSType.MOST_ONE) {
                 //QoS 0
                 sendPublish(sub.getClientId(), topic, qos, message, false);
@@ -263,6 +267,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         pubMessage.setTopicName(topic);
         pubMessage.setQos(qos);
         pubMessage.setPayload(message);
+        LOG.info(String.format("send publish message to <%s> on topic <%s>, message is [%s]", 
+                clientId, topic, new String(message)));
         if (pubMessage.getQos() != AbstractMessage.QOSType.MOST_ONE) {
             pubMessage.setMessageID(messageID);
         }
@@ -377,7 +383,17 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         //de-activate the subscriptions for this ClientID
 //        String clientID = (String) evt.getSession().getAttribute(Constants.ATTR_CLIENTID);
         subscriptions.deactivate(clientID);
-        LOG.info(String.format("Disconnected client <%s>", clientID));
+        LOG.info(String.format("Disconnected client <%s> with clean session %s", clientID, cleanSession));
+    }
+    
+    void proccessConnectionLost(String clientID) {
+        //If already removed a disconnect message was already processed for this clientID
+        if (m_clientIDs.remove(clientID) != null) {
+
+            //de-activate the subscriptions for this ClientID
+            subscriptions.deactivate(clientID);
+            LOG.info(String.format("Lost connection with client <%s>", clientID));
+        }
     }
     
     /**

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/SimpleMessaging.java

@@ -82,6 +82,10 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     public void disconnect(ServerChannel session) {
         disruptorPublish(new DisconnectEvent(session));
     }
+    
+    public void lostConnection(String clientID) {
+        disruptorPublish(new LostConnectionEvent(clientID));
+    }
 
     public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) {
         disruptorPublish(new ProtocolEvent(session, msg));
@@ -153,6 +157,9 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
 
         } else if (evt instanceof InitEvent) {
             processInit();
+        } else if (evt instanceof LostConnectionEvent) {
+            LostConnectionEvent lostEvt = (LostConnectionEvent) evt;
+            m_processor.proccessConnectionLost(lostEvt.getClientID());
         }
     }
 

broker/src/main/java/org/dna/mqtt/moquette/messaging/spi/impl/events/LostConnectionEvent.java

@@ -0,0 +1,19 @@
+package org.dna.mqtt.moquette.messaging.spi.impl.events;
+
+/**
+ * Used to model the connection lost event
+ * 
+ * @author andrea
+ */
+public class LostConnectionEvent extends MessagingEvent{
+    private String m_clientID;
+
+    public LostConnectionEvent(String clienID) {
+        m_clientID = clienID;
+    }
+
+    public String getClientID() {
+        return m_clientID;
+    }
+    
+}

broker/src/main/java/org/dna/mqtt/moquette/server/netty/NettyMQTTHandler.java

@@ -10,6 +10,7 @@ import org.dna.mqtt.moquette.proto.Utils;
 import org.dna.mqtt.moquette.proto.messages.AbstractMessage;
 import static org.dna.mqtt.moquette.proto.messages.AbstractMessage.*;
 import org.dna.mqtt.moquette.proto.messages.PingRespMessage;
+import org.dna.mqtt.moquette.server.Constants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -60,6 +61,9 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
     
     @Override
     public void channelInactive(ChannelHandlerContext ctx)throws Exception {
+        NettyChannel channel = m_channelMapper.get(ctx);
+        String clientID = (String) channel.getAttribute(Constants.ATTR_CLIENTID);
+        m_messaging.lostConnection(clientID);
         ctx.close(/*false*/);
         synchronized(m_channelMapper) {
             m_channelMapper.remove(ctx);