package com.wecloud.im.tillo.app_ws.service.impl;

import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.wecloud.im.tillo.app_ws.cache.UserCache;
import com.wecloud.im.tillo.app_ws.model.Constants;
import com.wecloud.im.tillo.app_ws.service.AppUserChannelsService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Description 维护netty用户channel对象
 * @Author hewei hwei1233@163.com
 * @Date 2019-12-03
 */
@Component
@Slf4j
public class AppUserChannelsServiceImpl implements AppUserChannelsService {

    private final static ThreadFactory NAMED_THREAD_FACTORY = new ThreadFactoryBuilder()
            .setNamePrefix("rpcWrite-").build();
//    .setPriority(3)
    /**
     * 远程调用ws下发数据线程池
     * 属于IO密集型任务
     * IO密集型任务线程并不是一直在执行任务，则应配置尽可能多的线程，如CPU核数*2
     * 某大厂设置策略：IO密集型时，大部分线程都阻塞，故需要多配置线程数： 公式：CPU核数/1-阻塞系数 阻塞系数：0.8-0.9
     * 比如8核CPU： 8/1-0.9 = 80 个线程数
     * <p>
     * 由于来自远程端调用下发数据  如果是群聊1000人群则调用1000次 为不要占用太多资源 需要排队下发
     * 经过并发测试  200并发1000人群消息  需要调用200x1000=20w次 考虑单机cpu性能还要顾及本机api业务 设置阻塞队列
     * 为避免过多占用本地io线程导致response慢，设置LinkedBlockingQueue数量多可以避免抢占，TODO （队列数量需要测试调试到最优数量 ）
     * 最大线程数量不要设置太多 数量、优先级一定要比本地io线程低级
     *
     *
     * <p>
     * TODO 待完善：消息发送投递至MQ， 消费方从MQ队列获取下发任务 本地队列不宜缓存太多(机器死机则会全丢失) 堆积的请求处理队列可能会耗费非常大的内存甚至死机
     */
//    private final static ExecutorService THREAD_POOL_RPC_WRITE_EXECUTOR =
//            new ThreadPoolExecutor(Constants.CPU_PROCESSORS * 4, Constants.CPU_PROCESSORS * 4 + 1,
//                    1L, TimeUnit.MILLISECONDS,
//                    new LinkedBlockingQueue<Runnable>(), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.CallerRunsPolicy());


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

    private final static ExecutorService THREAD_POOL_RPC_WRITE_EXECUTOR =
            new ThreadPoolExecutor(Constants.CPU_PROCESSORS * 100, Constants.CPU_PROCESSORS * 100,
                    1L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>(), NAMED_THREAD_FACTORY, new ThreadPoolExecutor.CallerRunsPolicy());


    @Autowired
    private UserCache userCache;


    @Override
    public NioSocketChannel get(String userId) {

        return AppUserChannelsService.CHANNEL_MAP.get(userId);
    }

    @Override
    public void put(String userId, NioSocketChannel channel) {

//        AppHashValueModel appHashValueModel = new AppHashValueModel();
//        appHashValueModel.setOnlineStatus(UserCache.ONLINE);
        userCache.online(userId);
        AppUserChannelsService.CHANNEL_MAP.put(userId, channel);


    }

    @Override
    public void remove(ChannelHandlerContext channelHandlerContext) {
        Channel channel = channelHandlerContext.channel();

        /*
            channel != null : 通道不能为空
            !channel.isActive() 通道不能是活跃状态的
            !channel.isOpen() 通道不能是打开状态的
           // !channel.isWritable() 通道是否可写数据
         */
        if (channel != null && !channel.isActive() && !channel.isOpen()) {
            String userId = String.valueOf(this.getUserIdByChannel(channelHandlerContext));
            userCache.offline(userId);


            log.debug("不活跃remove,uid:" + userId);
            AppUserChannelsService.CHANNEL_MAP.remove(userId);
            channelHandlerContext.channel().close();
        }
    }

    @Override
    public Long getUserIdByChannel(ChannelHandlerContext channelHandlerContext) {
        return channelHandlerContext.channel().attr(AppUserChannelsService.USER_ID).get();
    }

    @Override
    public Boolean isOnLocal(Long userId) {
        NioSocketChannel nioSocketChannel = this.get(String.valueOf(userId));
        return null != nioSocketChannel;

    }


    /**
     * TODO 待完成: 根据ACK回执 以及线程等待超时机制来判断客户端是否离线和超时;
     * TODO 待完成: 发送后阻塞当前子线程2秒后获取ack回执  如客户端发起ack回执则需要主动唤醒当前子线程 立马唤醒当前子线程， 判断如果已回执则返回发送成功， 如果未回执则判断客户端是否断线或发送错误
     *
     * @param msg
     * @param userId
     * @return
     */
    @Override
    public boolean rpcWriteData(String msg, Long userId) {

        Future<Boolean> future = THREAD_POOL_RPC_WRITE_EXECUTOR.submit(() -> {

            NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
            if (null == nioSocketChannel) {
                userCache.offline(String.valueOf(userId));
                log.debug("rpc-writeData连接为空:" + userId + "," + msg);
                return false;
            }

            // 判断连接是否断开
            if (nioSocketChannel.isShutdown()) {
                log.debug("rpc-writeData连接断开:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
                return false;
            }

            if (log.isInfoEnabled()) {
                log.info("rpc-writeData:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
            }

            ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msg));
            channelFuture.addListener(
                    //执行后回调的方法
                    (ChannelFutureListener) channelFuture1 -> {
                        if (log.isInfoEnabled()) {
                            log.info("rpc-netty线程异步执行结果:" + channelFuture1.isDone() + ",业务执行结果:" + channelFuture1.isSuccess()
                                    + "；\nwriteData:" + userId + "," + msg + ",channelId:" + nioSocketChannel.id().asLongText());
                        }
                    });

            channelFuture.get();

            return true;
        });

        boolean resultStatus = false;
        try {
            resultStatus = future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        return resultStatus;

    }

    @Override
    public boolean rpcKickWriteData(String msg, Long userId) {
        Future<Boolean> future = THREAD_POOL_RPC_WRITE_EXECUTOR.submit(() -> {

            NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
            if (null == nioSocketChannel) {
                log.debug("rpc-kickWriteData连接为空:" + userId + "," + msg);
                return false;
            }

            // 判断连接是否断开
            if (nioSocketChannel.isShutdown()) {
                log.debug("rpc-kickWriteData连接断开:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
                nioSocketChannel.close();
                return false;
            }

            if (log.isDebugEnabled()) {
                log.debug("rpc-kickWriteData:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
            }

            ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msg));
            channelFuture.addListener(
                    //执行后回调的方法
                    (ChannelFutureListener) channelFuture1 -> {
                        if (log.isDebugEnabled()) {
                            log.debug("rpc-netty踢人线程异步执行结果:" + channelFuture1.isDone() + ",业务执行结果:" + channelFuture1.isSuccess()
                                    + "；\nkickWriteData:" + userId + "," + msg + ",channelId:" + nioSocketChannel.id().asLongText());
                        }
                    });

            channelFuture.get();
            // 关闭
            nioSocketChannel.close();
            return true;
        });

        boolean resultStatus = false;
        try {
            resultStatus = future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        return resultStatus;
    }

    @Override
    public boolean rpcCloseOldChannel(Long userId) {
        Future<Boolean> future = THREAD_POOL_RPC_WRITE_EXECUTOR.submit(() -> {
            NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
            if (null == nioSocketChannel) {
                log.debug("rpc-closeOldChannel连接为空:" + userId);
                return false;
            }
            // 关闭
            nioSocketChannel.close();
            return true;
        });
        boolean resultStatus = false;
        try {
            resultStatus = future.get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        return resultStatus;
    }

    @Override
    public boolean writeData(String msg, Long userId) {

//        Future<Boolean> future = THREAD_POOL_EXECUTOR.submit(() -> {

        NioSocketChannel nioSocketChannel = get(String.valueOf(userId));
        if (null == nioSocketChannel) {
            userCache.offline(String.valueOf(userId));
            log.debug("writeData连接为空:" + userId + "," + msg);
            return false;
        }

        // 判断连接是否断开
        if (nioSocketChannel.isShutdown()) {
            log.debug("writeData连接断开:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
            return false;
        }

        if (log.isDebugEnabled()) {
            log.debug("writeData:" + userId + "," + msg + ",\nchannelId:" + nioSocketChannel.id().asLongText());
        }

        ChannelFuture channelFuture = nioSocketChannel.writeAndFlush(new TextWebSocketFrame(msg));
        channelFuture.addListener(
                //执行后回调的方法
                (ChannelFutureListener) channelFuture1 -> {
                    if (log.isDebugEnabled()) {
                        log.debug("netty线程异步执行结果:" + channelFuture1.isDone() + ",业务执行结果:" + channelFuture1.isSuccess()
                                + "；\nwriteData:" + userId + "," + msg + ",channelId:" + nioSocketChannel.id().asLongText());
                    }
                });

//            channelFuture.get();

        return true;
//        });

//        Boolean resultStatus = false;
//        try {
//            resultStatus = future.get();
//        } catch (InterruptedException | ExecutionException e) {
//            e.printStackTrace();
//        }
//        return resultStatus;

    }

}
