diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index 24e3f094..f1cf944a 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java @@ -13,6 +13,7 @@ * * You may elect to redistribute this code under either of these licenses. */ + package io.moquette.server.netty; import io.moquette.spi.impl.ProtocolProcessor; @@ -22,17 +23,15 @@ import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.codec.mqtt.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; - import static io.netty.handler.codec.mqtt.MqttQoS.AT_MOST_ONCE; - /** * * @author andrea */ @Sharable public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { - + private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class); private final ProtocolProcessor m_processor; @@ -75,22 +74,27 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { m_processor.processPubAck(ctx.channel(), (MqttPubAckMessage) msg); break; case PINGREQ: - MqttFixedHeader pingHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, AT_MOST_ONCE, false, 0); + MqttFixedHeader pingHeader = new MqttFixedHeader( + MqttMessageType.PINGRESP, + false, + AT_MOST_ONCE, + false, + 0); MqttMessage pingResp = new MqttMessage(pingHeader); ctx.writeAndFlush(pingResp); break; } } catch (Throwable ex) { - LOG.error("Exception was caught while processing MQTT message, " + ex.getCause(), ex); - ctx.fireExceptionCaught(ex); + LOG.error("Exception was caught while processing MQTT message, " + ex.getCause(), ex); + ctx.fireExceptionCaught(ex); } } - + @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { String clientID = NettyUtils.clientID(ctx.channel()); if (clientID != null && !clientID.isEmpty()) { - LOG.info("Notifying connection lost event. MqttClientId = {}.", clientID); + LOG.info("Notifying connection lost event. MqttClientId = {}.", clientID); m_processor.processConnectionLost(clientID, ctx.channel()); } ctx.close(); @@ -98,9 +102,12 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { - LOG.error( - "An unexpected exception was caught while processing MQTT message. Closing Netty channel. MqttClientId = {}, cause = {}, errorMessage = {}.", - NettyUtils.clientID(ctx.channel()), cause.getCause(), cause.getMessage()); + LOG.error( + "An unexpected exception was caught while processing MQTT message. " + + "Closing Netty channel. MqttClientId = {}, cause = {}, errorMessage = {}.", + NettyUtils.clientID(ctx.channel()), + cause.getCause(), + cause.getMessage()); ctx.close(); }