package tillo.app_ws;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import tillo.app_ws.model.Constants;
import tillo.app_ws.receive.ReadWsData;
import tillo.app_ws.service.AppUserChannelsService;

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 AppImHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    @Resource
    private ReadWsData readWsData;
    @Resource
    private AppUserChannelsService appUserChannelsService;

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

    private final static ExecutorService THREAD_POOL_EXECUTOR =
//            new ThreadPoolExecutor(6, 10,
//                    0L, TimeUnit.MILLISECONDS,
//                    new LinkedBlockingQueue<Runnable>(2048), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.AbortPolicy());

            new ThreadPoolExecutor(Constants.CPU_PROCESSORS * 120, Constants.CPU_PROCESSORS * 130 + 2,
                    1L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>(), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.CallerRunsPolicy());

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

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

    private void execute(ChannelHandlerContext ctx, String data) {
        Long userIdByChannel = appUserChannelsService.getUserIdByChannel(ctx);

        log.debug("appWS收到" + userIdByChannel + ":" + data + ",channelId:" + ctx.channel().id().asLongText());

        String language = ctx.channel().attr(AppUserChannelsService.LANGUAGE).get();
        readWsData.convertModel(data, userIdByChannel, language);
    }


    /**
     * 检测到异常
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

        //排除当客户端意外关闭的情况，不是发送指定指令通知服务器退出，就会产生此错误。
        if (ctx.channel().isActive()) {
            Long userIdByChannel = appUserChannelsService.getUserIdByChannel(ctx);
            log.error("uid:" + userIdByChannel + ",ws异常,channelId:" + ctx.channel().id().asLongText(), cause);
        }
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        Long userIdByChannel = appUserChannelsService.getUserIdByChannel(ctx);

        log.debug("uid:" + userIdByChannel + "，app端连接WS成功" + ",channelId:" + ctx.channel().id().asLongText());

    }

    /**
     * 客户端不活跃
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        Long userIdByChannel = appUserChannelsService.getUserIdByChannel(ctx);
        log.debug("uid:" + userIdByChannel + "," + "不活跃" + ",channelId:" + ctx.channel().id().asLongText());

        appUserChannelsService.remove(ctx);
    }

    /**
     * 移除时触发， 不活跃的情况下会移除，会再次触发该事件
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        Long userIdByChannel = appUserChannelsService.getUserIdByChannel(ctx);
        log.debug("uid:" + userIdByChannel + "," + "handlerRemoved" + ",channelId:" + ctx.channel().id().asLongText());
    }
}

