diff --git a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java b/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java index 318647e8..4367e688 100644 --- a/broker/src/main/java/io/moquette/server/netty/metrics/MQTTMessageLogger.java +++ b/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 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 ee4d7ee4..a2fca7b2 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/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)); } }