package chat.server.moquette;

import org.apache.log4j.Logger;

import chat.server.ServerInstance;
import chat.server.im.IMDispatcher;
import chat.server.moquette.message.ClientID;
import chat.server.moquette.message.MqttConnAckMessage;
import chat.server.moquette.message.MqttConnAckVariableHeader;
import chat.server.moquette.message.MqttConnectAckPayload;
import chat.server.moquette.message.MqttConnectMessage;
import chat.server.moquette.message.MqttConnectPayload;
import chat.server.moquette.message.MqttConnectReturnCode;
import chat.server.moquette.message.MqttConnectVariableHeader;
import chat.server.moquette.message.MqttFixedHeader;
import chat.server.moquette.message.MqttMessage;
import chat.server.moquette.message.MqttMessageType;
import chat.server.moquette.message.MqttPublishMessage;
import chat.server.moquette.message.MqttQoS;
import chat.user.Session;
import chat.user.SessionStore;
import cn.wildfirechat.proto.WFCMessage;
import cn.wildfirechat.proto.WFCMessage.ConnectAckPayload;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.AttributeKey;
import io.netty.util.ReferenceCountUtil;

public class MqttDispatcher extends ChannelInboundHandlerAdapter {

    private static AttributeKey<Object> ATTR_KEEPALIVE = AttributeKey.valueOf("keepAlive");
    private static AttributeKey<Object> ATTR_CLEANSESSION = AttributeKey.valueOf("cleanSession");
    private static AttributeKey<Object> ATTR_CLIENTID = AttributeKey.valueOf("ClientID");

	private static Logger logger;
	private static IMDispatcher imDispatcher;
	
	static {
		logger = Logger.getLogger(MqttDispatcher.class);
		imDispatcher = new IMDispatcher();
	}
	
	public MqttDispatcher() {
		
	}
	
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object message) {
        try {
            MqttMessage msg = (MqttMessage) message;
            MqttMessageType messageType = msg.messageType();
            
            logger.info("Processing MQTT message, type=" + messageType);
            
            Channel channel = ctx.channel();

            switch (messageType) {
                case CONNECT:
                    processConnect(channel, msg);
                    break;
                case DISCONNECT:
                    processDisconnect(channel, msg);
                    break;
                case PUBLISH:
                	MqttPublishMessage mess = (MqttPublishMessage) msg;
                	String topic = mess.variableHeader().topicName();
                	if (topic.equals("MP")) {
                		System.out.println("MP");
                	}
                    processPublish(channel, msg);
                    break;
                case PINGREQ:
                	processPing(channel, msg);
                    break;
                default:
                	logger.error("Unkonwn MessageType:" + messageType);
                    break;
            }
        } 
        catch (Throwable ex) {
        	logger.error("Exception was caught while processing MQTT message, " + ex.getCause(), ex);
            ctx.fireExceptionCaught(ex);
            ctx.close();
        } 
        finally {
            ReferenceCountUtil.release(message);
        }
    }
    
	public void processConnect(Channel channel, MqttMessage msg) {
    	if (logger.isDebugEnabled()) {
            logger.debug("receive connect(channel=)" + channel.id());
    	}
    	
		MqttConnectMessage message = (MqttConnectMessage) msg;
        MqttConnectPayload payload = message.payload();
        
        ClientID clientID = ClientID.valueOf(payload);

        //1. if starting or terminated
        if (!ServerInstance.isStart()) {
            channel.close();
            return;
        }
        
        //2. if empty client id
        if (clientID.isEmpty()) {
            MqttConnAckMessage badIdMessage = createConnAckMessageFalse(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, null);
            channel.writeAndFlush(badIdMessage);
            channel.close();
            return;
        }
        
        //3. set channel attribute
        MqttConnectVariableHeader header = message.variableHeader();
        customizeChannel(channel, clientID, header);
        
        //4. session
        clientID.setValueTo(channel);
        
        Session session = SessionStore.get(clientID);
        
        if (session != null) {
            session.setChannel(channel);
            session.refreshLastActiveTime();
        }
        
        //5. write ACK message
        ConnectAckPayload ackPayload = createConnAckPayload(session);
        MqttConnAckMessage successMessage = createConnAckMessage(MqttConnectReturnCode.CONNECTION_ACCEPTED, ackPayload.toByteArray());
        channel.writeAndFlush(successMessage);
        
        //6. notify user state
//        User user = session.getUser();
//        user.notify(PublishedOperator.Online);
    }
    
    private ConnectAckPayload createConnAckPayload(Session session) {
    	long timestamp = System.currentTimeMillis();
        long messageHead = timestamp; //session.getMessageHead();
        long friendHead = timestamp;//session.getFriendHead();
        long friendRqHead = timestamp;//session.getFriendRqHead();
        long settingHead = timestamp;//session.getSettingHead();
        
        ConnectAckPayload.Builder builder = WFCMessage.ConnectAckPayload.newBuilder();
        builder.setMsgHead(messageHead);
        builder.setFriendHead(friendHead);
        builder.setFriendRqHead(friendRqHead);
        builder.setSettingHead(settingHead);
        builder.setServerTime(System.currentTimeMillis());     

        ConnectAckPayload payload = builder.build();
        return payload;
    }
    
	private MqttConnAckMessage createConnAckMessage(MqttConnectReturnCode code, byte[] data) {
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(code, true);
        
        MqttConnAckMessage result = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader, new MqttConnectAckPayload(data));
		return result;
	}
	
	private MqttConnAckMessage createConnAckMessageFalse(MqttConnectReturnCode code, byte[] data) {
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(code, false);
        
        MqttConnAckMessage result = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader, new MqttConnectAckPayload(data));
		return result;
	}	
    
	private void customizeChannel(Channel channel, ClientID clientID, MqttConnectVariableHeader header) {
		//1. set clientID
		channel.attr(ATTR_CLIENTID).set(clientID.getValue());
		
		//2. live time
        int keepAliveTimeSeconds = header.keepAliveTimeSeconds();
		channel.attr(ATTR_KEEPALIVE).set(keepAliveTimeSeconds);
		
		//3.  clean session
		boolean cleanSession = header.isCleanSession();
		channel.attr(ATTR_CLEANSESSION).set(cleanSession);

		//4. idle time
		ChannelPipeline pipeline = channel.pipeline();
		
        if (pipeline.names().contains("idleStateHandler")) {
            pipeline.remove("idleStateHandler");
        }
        
        int idleTime = Math.round(keepAliveTimeSeconds * 1.5f);
        pipeline.addFirst("idleStateHandler", new IdleStateHandler(idleTime, 0, 0));
	}

	public void processDisconnect(Channel channel, MqttMessage msg) throws InterruptedException {
    	if (logger.isDebugEnabled()) {
            logger.debug("receive disconnect(channel=)" + channel.id());
    	}
    	
    	ClientID clientID = ClientID.valueOf(channel);
    	
    	//1. flush
    	channel.flush();

    	//2. if empty
        if (clientID.isEmpty()) {
            channel.close();
            return;
        }

        //3. close session channel
        Session session = SessionStore.get(clientID);
        
        if (session == null) {
        	return;
        }
        
        session.refreshLastActiveTime();
        session.closeChannel();
        
        //4. delete session
        SessionStore.delete(clientID);
    }
    
    public void processPublish(Channel channel, MqttMessage message) throws Exception {
    	if (logger.isDebugEnabled()) {
    		logger.debug("receive request(channel=)" + channel.id() + ": publish");
    	}
    	
    	MqttPublishMessage publishMessage = (MqttPublishMessage) message;
    	imDispatcher.channelRead0(channel, publishMessage);
    }
    
    private void processPing(Channel channel, MqttMessage message) {
    	if (logger.isDebugEnabled()) {
        	logger.debug("receive ping(channel=)" + channel.id());
    	}
    	
        MqttFixedHeader pingHeader = new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0);
        MqttMessage result = new MqttMessage(pingHeader);
        channel.writeAndFlush(result);		
	}
    
    

}