package com.docus.server.common.netty.server.handler;

import com.docus.core.util.Func;
import com.docus.core.util.json.JSON;
import com.docus.server.common.CommMsg;
import com.docus.server.common.netty.server.TerminalToChannelCacheMap;
import com.docus.server.dto.scheduling.management.schterminator.NettyTerminatorDTO;
import com.docus.server.enums.BusyStateEnum;
import com.docus.server.enums.OnlineStateEnum;
import com.docus.server.vo.scheduling.management.schcollector.LoadSchCollectorVO;
import com.fasterxml.jackson.core.type.TypeReference;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.Serializable;
import java.net.InetSocketAddress;
import java.util.List;

import static com.docus.server.common.MsgConstants.*;

/**
 * 业务消息处理
 */
@Slf4j
@ChannelHandler.Sharable
@Component
public class NettyBusinessHandler extends SimpleChannelInboundHandler<CommMsg<Serializable>> {

    private static final ChannelGroup DEFAULT_CHANNEL_GROUP = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Resource
    private TerminalToChannelCacheMap terminalToChannelCacheMap;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, CommMsg commMsg) throws Exception {

        String messageType = commMsg.getMessageType();
        String messageTime = commMsg.getMessageTime();
        String messageContent = (String) commMsg.getContent();

        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = ipSocket.getAddress().getHostAddress();

        log.info("======== 【调度器服务端-收到终端消息】, 消息时间={}, 消息类型={}, 消息内容={}", messageTime, messageType, messageContent + " ======== ");

        switch (messageType) {
            case ONLINE_REGISTER:
                onlineRegister(ctx, messageContent, clientIp);
                break;
            case TERMINATOR_RESTART:
                terminatorRestart(messageContent);
                break;
            case COLLECTOR_RESTART:
                collectorRestart(messageContent);
                break;
            case VIRTUAL_RESTART:
                virtualRestart(messageContent);
                break;
            case UPDATE_COLLECTOR_FILE:
                updateCollectorFile(messageContent);
                break;
            case UPDATE_COLLECTOR_CONFIG:
                updateCollectorConfig(messageContent);
                break;
            case SCH_DISTRIBUTE_TASKS:
                schDistributeTasks(commMsg, messageContent);
                break;
            case HAS_COLLECTOR_COUNT:
                hasCollectorCount(ctx, messageContent);
                break;
            default:
                break;
        }
    }

    private void hasCollectorCount(ChannelHandlerContext ctx, String messageContent) {
        log.info("【调度器服务端-收到终端消息】收到终端客户端返回的【采集器数量】消息，内容{}", messageContent);

        String terminalIp = terminalToChannelCacheMap.getClientKey(ctx.channel());

        NettyTerminatorDTO nettyTerminatorDTO = terminalToChannelCacheMap.getTerminatorByIp(terminalIp);

        if (Func.isNotBlank(messageContent)) {
            nettyTerminatorDTO.getLoadSchCollectorVOList().clear();
            List<LoadSchCollectorVO> loadSchCollectorVOList = JSON.fromJSONWithGeneric(messageContent, new TypeReference<List<LoadSchCollectorVO>>() {
            });
            nettyTerminatorDTO.getLoadSchCollectorVOList().addAll(loadSchCollectorVOList);
        }
    }

    private void schDistributeTasks(CommMsg commMsg, String messageContent) {
        log.info("【调度器服务端-收到终端消息】收到采集调度器下发任务命令，内容{}", messageContent);
    }

    private void updateCollectorConfig(String messageContent) {
        log.info("【调度器服务端-收到终端消息】收到更新采集器配置命令，内容{}", messageContent);
    }

    private void updateCollectorFile(String messageContent) {
        log.info("【调度器服务端-收到终端消息】收到更新采集器文件命令，内容{}", messageContent);
    }

    private void virtualRestart(String messageContent) {
        log.info("【调度器服务端-收到终端消息】收到虚拟机重启命令，内容{}", messageContent);
    }

    private void collectorRestart(String messageContent) {
        log.info("【调度器服务端-收到终端消息】收到采集器重启命令，内容{}", messageContent);
    }

    private void terminatorRestart(String messageContent) {

        log.info("【调度器服务端-收到终端消息】接受到终端重启命令，内容{}", messageContent);
    }

    private void onlineRegister(ChannelHandlerContext ctx, String messageContent, String clientIp) {
        log.info("【调度器服务端-收到终端消息】接受到终端注册命令，内容{}", messageContent);
        String clientKey = terminalToChannelCacheMap.getClientKey(ctx.channel());

        if (Func.isNotBlank(clientKey)) {

            NettyTerminatorDTO nettyTerminatorDTO = new NettyTerminatorDTO();
            nettyTerminatorDTO.setTerminatorIp(clientIp);
            nettyTerminatorDTO.setBusyState(BusyStateEnum.IDLE);
            nettyTerminatorDTO.setOnlineState(OnlineStateEnum.ONLINE);

            //将ip和channel进行映射
            terminalToChannelCacheMap.put(nettyTerminatorDTO, ctx.channel());
        }
    }

    /**
     * netty client 上线
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelRegistered();

        InetSocketAddress ipSocket = (InetSocketAddress) ctx.channel().remoteAddress();
        String clientIp = ipSocket.getAddress().getHostAddress();
        log.info("【采集器-终端IP】:{}，连接上线,IP地址信息:{}", clientIp, clientIp);

        String clientId = terminalToChannelCacheMap.getClientKey(ctx.channel());

        if (Func.isBlank(clientId)) {

            NettyTerminatorDTO nettyTerminatorDTO = new NettyTerminatorDTO();
            nettyTerminatorDTO.setTerminatorIp(clientIp);
            nettyTerminatorDTO.setBusyState(BusyStateEnum.IDLE);
            nettyTerminatorDTO.setOnlineState(OnlineStateEnum.OFFLINE);

            terminalToChannelCacheMap.put(nettyTerminatorDTO, ctx.channel());
        }

        DEFAULT_CHANNEL_GROUP.add(ctx.channel());

        System.out.println(ctx.channel().remoteAddress() + " 上线," + "【调度器-终端】在线数量：" + DEFAULT_CHANNEL_GROUP.size());
    }

    /**
     * netty client 下线
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        System.out.println(channel.remoteAddress() + " 下线," + "【调度器-终端】在线数量：" + DEFAULT_CHANNEL_GROUP.size());
        String clientId = terminalToChannelCacheMap.getClientKey(channel);
        log.error("客户端下线，终端连接：{}", clientId);

        //移除终端，终端离线
        if (clientId != null) {
            terminalToChannelCacheMap.remove(clientId);
        }
    }

    /**
     * netty exception 通道异常
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        Channel channel = ctx.channel();
        if (channel.isActive()) {
            ctx.close();
        }

        super.exceptionCaught(ctx, cause);
    }

}
