/*
 * Copyright (c) 2012-2017 The original author or authorsgetRockQuestions()
 * ------------------------------------------------------
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * and Apache License v2.0 which accompanies this distribution.
 *
 * The Eclipse Public License is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * The Apache License v2.0 is available at
 * http://www.opensource.org/licenses/apache2.0.php
 *
 * You may elect to redistribute this code under either of these licenses.
 */
package io.moquette.server.netty;

import io.moquette.parser.proto.Utils;
import io.moquette.parser.proto.messages.*;
import io.moquette.spi.impl.ProtocolProcessor;
import static io.moquette.parser.proto.messages.AbstractMessage.*;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import io.netty.handler.codec.CorruptedFrameException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;

/**
 *
 * @author andrea
 */
@Sharable
public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
    
    private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);
    private final ProtocolProcessor m_processor;

    public NettyMQTTHandler(ProtocolProcessor processor) {
        m_processor = processor;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object message) {
        AbstractMessage msg = (AbstractMessage) message;
        LOG.info("Received a message of type {}", Utils.msgType2String(msg.getMessageType()));
        try {
            switch (msg.getMessageType()) {
                case CONNECT:
                    m_processor.processConnect(ctx.channel(), (ConnectMessage) msg);
                    break;
                case SUBSCRIBE:
                    m_processor.processSubscribe(ctx.channel(), (SubscribeMessage) msg);
                    break;
                case UNSUBSCRIBE:
                    m_processor.processUnsubscribe(ctx.channel(), (UnsubscribeMessage) msg);
                    break;
                case PUBLISH:
                    m_processor.processPublish(ctx.channel(), (PublishMessage) msg);
                    break;
                case PUBREC:
                    m_processor.processPubRec(ctx.channel(), (PubRecMessage) msg);
                    break;
                case PUBCOMP:
                    m_processor.processPubComp(ctx.channel(), (PubCompMessage) msg);
                    break;
                case PUBREL:
                    m_processor.processPubRel(ctx.channel(), (PubRelMessage) msg);
                    break;
                case DISCONNECT:
                    m_processor.processDisconnect(ctx.channel());
                    break;
                case PUBACK:
                    m_processor.processPubAck(ctx.channel(), (PubAckMessage) msg);
                    break;
                case PINGREQ:
                    PingRespMessage pingResp = new PingRespMessage();
                    ctx.writeAndFlush(pingResp);
                    break;
            }
        } catch (Exception ex) {
            LOG.error("Bad error in processing the message", ex);
            ctx.fireExceptionCaught(ex);
        }
    }
    
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        String clientID = NettyUtils.clientID(ctx.channel());
        if (clientID != null && !clientID.isEmpty()) {
            m_processor.processConnectionLost(clientID, ctx.channel());
        }
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        if (cause instanceof CorruptedFrameException) {
            //something goes bad with decoding
            LOG.warn("Error decoding a packet, probably a bad formatted packet, message: " + cause.getMessage());
        } else if (cause instanceof IOException && "Connection reset by peer".equals(cause.getMessage())) {
            LOG.warn("Network connection closed abruptly");
        } else {
            LOG.error("Ugly error on networking", cause);
        }
        ctx.close();
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        if (ctx.channel().isWritable()) {
            m_processor.notifyChannelWritable(ctx.channel());
        }
        ctx.fireChannelWritabilityChanged();
    }

}

@@ -996,63 +996,8 @@
 r;%0A%0A
-import io.netty.handler.codec.CorruptedFrameException;%0A
 %0Aimp
@@ -1055,36 +1055,8 @@
 y;%0A%0A
-import java.io.IOException;%0A
 %0A/**
@@ -1557,49 +1557,28 @@
     
-LOG.info(%22Received a message of type %7B%7D%22,
+String messageType =
  Uti
@@ -1616,18 +1616,142 @@
 eType())
-)
 ;
+%0A        if (LOG.isDebugEnabled()) %7B%0A        %09LOG.debug(%22Processing MQTT message. MessageType = %7B%7D.%22, messageType);%0A        %7D
 %0A       
@@ -3283,59 +3283,84 @@
 ch (
-Exception ex) %7B%0A            LOG.error(%22Bad error in
+Throwable ex) %7B%0A%09%09%09LOG.error(%0A%09%09%09%09%09%22An unexpected exception was caught while
  pro
@@ -3371,19 +3371,20 @@
 ing 
-the
+MQTT
  message
 %22, e
@@ -3383,28 +3383,115 @@
 sage
-%22, ex);%0A            
+. MessageType = %7B%7D, cause = %7B%7D, errorMessage = %7B%7D.%22,%0A%09%09%09%09%09messageType, ex.getCause(), ex.getMessage());%0A%09%09%09
 ctx.
@@ -3741,24 +3741,102 @@
 sEmpty()) %7B%0A
+%09%09%09LOG.info(%22Notifying connection lost event. MqttClientId = %7B%7D.%22, clientID);%0A
             
@@ -4046,477 +4046,245 @@
 n %7B%0A
-        if (cause instanceof CorruptedFrameException) %7B%0A            //something goes bad with decoding%0A            LOG.warn(%22Error decoding a packet, probably a bad formatted packet, message: %22 + cause.getMessage());%0A        %7D else if (cause instanceof IOException && %22Connection reset by peer%22.equals(cause.getMessage())) %7B%0A            LOG.warn(%22Network connection closed abruptly%22);%0A        %7D else %7B%0A            LOG.error(%22Ugly error on networking%22, cause);%0A        %7D
+%09%09LOG.error(%0A%09%09%09%09%22An unexpected exception was caught while processing MQTT message. Closing Netty channel. MqttClientId = %7B%7D, cause = %7B%7D, errorMessage = %7B%7D.%22,%0A%09%09%09%09NettyUtils.clientID(ctx.channel()), cause.getCause(), cause.getMessage());
 %0A   

Improve logging: - When a MQTT message is received, we should only log it in debug level. It is better to log the actions performed in the protocol processor at the info level (they are more meaningful). - Capture and log every exception raised while processing the message (including RuntimeExceptions).