package com.wecloud.im.netty.core;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.wecloud.im.ws.model.WsConstants;
import com.wecloud.im.ws.receive.ReadWsData;
import com.wecloud.im.ws.service.MangerChannelService;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Description app端 长连接事件处理
 * @Author hewei hwei1233@163.com
 * @Date 2019-07-26
 */
@Component
@ChannelHandler.Sharable
@Slf4j
public class WsReadHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    private static final String PING = "ping";
    private static final String PONG = "pong";

    @Resource
    private ReadWsData readWsData;

    @Resource
    private MangerChannelService mangerChannelService;

    private final static ThreadFactory NAMED_THREAD_FACTORY = new ThreadFactoryBuilder()
            .setNamePrefix("WS-business-").build();
    /**
     * 耗时核心业务处理线程池
     * 属于io密集型业务
     * io密集型任务配置尽可能多的线程数量
     */
    private final static ExecutorService TASK_THREAD_POOL_EXECUTOR =
            new ThreadPoolExecutor(WsConstants.CPU_PROCESSORS * 5, WsConstants.CPU_PROCESSORS * 10,
                    10L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>(10), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.CallerRunsPolicy());

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
        ctx.channel().attr(MangerChannelService.READ_IDLE_TIMES).set(0);// 读空闲的计数清零

        String data = msg.text();
        try {
            if (data.isEmpty()) {
                return;
            }
            /*
             * 在此进入耗时业务线程池，  将不再阻塞netty的I/O线程，提高网络吞吐
             */
            TASK_THREAD_POOL_EXECUTOR.execute(() ->
                    execute(ctx, data)
            );

        } catch (Exception e) {
            //返回错误
            ctx.channel().writeAndFlush(new TextWebSocketFrame("error=" + e + ",data=" + data));
            log.error(e.getMessage() + data, e);
        }
    }

    private void execute(ChannelHandlerContext ctx, String data) {
        String appKey = ctx.channel().attr(MangerChannelService.APP_KEY).get();
        String clientId = ctx.channel().attr(MangerChannelService.CLIENT_ID).get();
        try {

            if (PING.equals(data)) {
                log.info("收到心跳clientId:" + clientId);
                ctx.channel().writeAndFlush(new TextWebSocketFrame(PONG));
                return;
            }

            if (PONG.equals(data)) {
                log.info("收到心跳应用Pong,clientId:" + clientId);
                return;
            }

            readWsData.convertModel(data, ctx, appKey, clientId);
        } catch (Exception e) {
            log.error("系统繁忙data:" + data + ",appKey:" + appKey + ",clientId:" + clientId +
                    ",channelId:" + ctx.channel().id().asLongText(), e);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        String clientId = ctx.channel().attr(MangerChannelService.CLIENT_ID).get();

        //读超时计时器
        Integer readIdleTimes = ctx.channel().attr(MangerChannelService.READ_IDLE_TIMES).get();

        IdleStateEvent event = (IdleStateEvent) evt;

        String eventType = null;


        switch (event.state()) {
            case READER_IDLE:
                eventType = "读空闲:readIdleTimes=" + readIdleTimes;
                ctx.channel().attr(MangerChannelService.READ_IDLE_TIMES).set(readIdleTimes + 1);// 读空闲的计数加1

                // 发ping
                ctx.channel().writeAndFlush(new TextWebSocketFrame(PING));
                break;
            case WRITER_IDLE:
                eventType = "写空闲";
                // 不处理
                break;
            case ALL_IDLE:
                eventType = "读写空闲";
                // 不处理
                break;
        }
        log.info(clientId + "超时事件：" + eventType);
        if (readIdleTimes >= 5) {
            log.info(clientId + ".读空闲超过5次关闭连接");
            ctx.channel().close();
        }
    }


//    /**
//     * 检测到异常
//     *
//     * @param ctx
//     * @param cause
//     * @throws Exception
//     */
//    @Override
//    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
//        String userIdByChannel = mangerChannelService.getInfoByChannel(ctx);
//        log.info("uid:" + userIdByChannel + ",ws异常,channelId:" + ctx.channel().id().asLongText(), cause);
//    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        String userIdByChannel = mangerChannelService.getInfoByChannel(ctx);
        log.info("连接WS成功handlerAdded,uid:" + userIdByChannel + "，" + ",channelId:" + ctx.channel().id().asLongText());

    }

//    /**
//     * 客户端不活跃
//     *
//     * @param ctx
//     * @throws Exception
//     */
//    @Override
//    public void channelInactive(ChannelHandlerContext ctx) {
//        String userIdByChannel = mangerChannelService.getInfoByChannel(ctx);
//        log.info("uid:" + userIdByChannel + "," + "channelInactive" + ",channelId:" + ctx.channel().id().asLongText());
//    }

    /**
     * 移除时触发， 不活跃的情况下会移除，会再次触发该事件
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        String userIdByChannel = mangerChannelService.getInfoByChannel(ctx);
        log.info("uid:" + userIdByChannel + "," + "handlerRemoved" + ",channelId:" + ctx.channel().id().asLongText());
        // 关掉连接
        ctx.close();
    }
}
