Fixed inactive channel propagation event through MQTTMessageLogger

broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java

@@ -37,12 +37,15 @@ public class MQTTMessageLogger extends ChannelDuplexHandler {
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object message) {
-        logMQTTMessage(ctx, (AbstractMessage) message, "C->B");
+        logMQTTMessage(ctx, message, "C->B");
         ctx.fireChannelRead(message);
     }
 
-    private void logMQTTMessage(ChannelHandlerContext ctx, AbstractMessage message, String direction) {
-        AbstractMessage msg = message;
+    private void logMQTTMessage(ChannelHandlerContext ctx, Object message, String direction) {
+        if (!(message instanceof AbstractMessage)) {
+            return;
+        }
+        AbstractMessage msg = (AbstractMessage) message;
         String clientID = NettyUtils.clientID(ctx.channel());
         switch (msg.getMessageType()) {
             case CONNECT:
@@ -89,7 +92,7 @@ public class MQTTMessageLogger extends ChannelDuplexHandler {
         if (clientID != null && !clientID.isEmpty()) {
             LOG.info("Channel closed <{}>", clientID);
         }
-        ctx.fireChannelActive();
+        ctx.fireChannelInactive();
     }
 
     @Override

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java

@@ -385,12 +385,11 @@ public class ProtocolProcessor {
     }
 
     private void setupAutoFlusher(ChannelPipeline pipeline, int flushIntervalMs) {
-        AutoFlushHandler autoFlushHandler = new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS);
         try {
-            pipeline.addAfter("idleEventHandler", "autoFlusher", autoFlushHandler);
+            pipeline.addAfter("idleEventHandler", "autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
         } catch (NoSuchElementException nseex) {
             //the idleEventHandler is not present on the pipeline
-            pipeline.addFirst("autoFlusher", autoFlushHandler);
+            pipeline.addFirst("autoFlusher", new AutoFlushHandler(flushIntervalMs, TimeUnit.MILLISECONDS));
         }
     }