package com.jay.monitor.data.client;

import com.alipay.remoting.InvokeContext;
import com.alipay.remoting.rpc.RpcClient;
import com.jay.monitor.data.core.enums.MonitorDataConstants;
import com.jay.monitor.data.core.model.serializable.base.ApplicationData;
import com.jay.monitor.data.core.model.serializable.base.BaseDTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.LongAdder;

/**
 * 监控数据客户端传输列
 *
 * @author ： liukx
 * @time ： 2020/8/26 - 14:25
 */
public class MonitorDataClient implements Runnable {

    private LongAdder counter = new LongAdder();

    private Logger logger = LoggerFactory.getLogger(getClass());

    private String host;

    private Integer port = MonitorDataConstants.DEFAULT_PORT;

    private String addr;

    /**
     * 是否接收监控请求数据
     */
    private volatile boolean isEnabledReceive = true;

    private RpcClient client = new RpcClient();

    private ExecutorService executorService = Executors.newSingleThreadExecutor();

    private Integer queueSize = 3000;

    private Queue<BaseDTO> queueData;

    /**
     * 全局上下文传递信息，用于存储一些公共信息
     */
    private InvokeContext invokeContext = new InvokeContext();

    public MonitorDataClient(String host, Integer port) {
        this.host = host;
        this.port = port;
        this.addr = host + ":" + port;
    }

    /**
     * 初始化数据
     */
    public void init() {
        queueData = new ArrayBlockingQueue<>(getQueueSize());
        executorService.execute(this);
        client.startup();
    }

    public Integer getQueueSize() {
        return queueSize;
    }

    public void setQueueSize(Integer queueSize) {
        this.queueSize = queueSize;
    }

    public String getAddr() {
        return addr;
    }

    /**
     * 单向发送
     *
     * @param data
     * @throws Exception
     */
    private void sendOnWay(BaseDTO data) throws Exception {

        String applicationName = invokeContext.get(MonitorDataConstants.INVOKE_CLIENT_NAME).toString();
        String ip = invokeContext.get(MonitorDataConstants.INVOKE_CLIENT_IP).toString();

        if (data instanceof ApplicationData) {
            ApplicationData applicationData = (ApplicationData) data;
            if (applicationData.getApplicationName() == null) {
                applicationData.setApplicationName(applicationName);
            }
            if (applicationData.getIp() == null) {
                applicationData.setIp(ip);
            }
        }

        client.oneway(getAddr(), data, invokeContext);
        if (!isEnabledReceive()) {
            enabledReceive();
        }
    }

    public boolean isEnabledReceive() {
        return isEnabledReceive;
    }

    private void enabledReceive() {
        isEnabledReceive = true;
        logger.info("开启接收监控数据标识");
    }

    private void offReceive() {
        isEnabledReceive = false;
        logger.warn("关闭接收监控数据标识");
    }

    /**
     * 发送数据
     *
     * @param data
     */
    public void sendObjectMsg(BaseDTO data) {
        if (isEnabledReceive()) {
            boolean isAdd = queueData.offer(data);
            if (!isAdd) {
                counter.increment();
                long current = counter.longValue();
                if (current % 1000 == 0) {
                    logger.warn("容量不够，请及时扩容 queueSize:" + this.queueSize);
                    counter.reset();
                }
            }
        }
    }

    public void addInvokeContext(String key, String value) {
        invokeContext.put(key, value);
    }

    @Override
    public void run() {
        logger.info(" 开启监控数据传输线程 ... ");
        int threshold = this.queueSize / 2;
        int rate = this.queueSize / 10;
        while (true) {
            BaseDTO baseDTO = queueData.poll();
            try {
                if (baseDTO != null) {
                    sendOnWay(baseDTO);
                }

                int size = queueData.size();

                // 当未消费数量超过1半的时候，说明消费已经跟不上了。这时候尝试开始加速消费。
                if (size >= threshold) {
                    logger.warn("开启加速消费 当前数量:" + size);
                    // 加速消费，按照当前总数10%开始发送，尽可能不占用系统资源。
                    speedConsumer(rate);
                }
            } catch (Exception e) {
                logger.warn("monitor consumer fail : " + e.getMessage());
                // 当遇到连接问题的时候,尝试休眠等待.
                connectionProblem(e);
            } finally {
                sleep(5);
            }

        }
    }

    /**
     * 连接问题处理
     *
     * @param e
     */
    private void connectionProblem(Exception e) {
        if (e.getMessage().startsWith("Create connection failed.")) {
            // 如果连接不上了，先关闭接收数据入口，休眠一段时间
            offReceive();
            sleep(30000);
            enabledReceive();
        }
        // 消费不过来了.
        else if (e.getMessage().startsWith("Check connection failed for address")) {
            // offReceive();
            sleep(5000);
            // enabledReceive();
        }
    }

    /**
     * 遇到一些不可控的情况，尝试休眠，让出CPU
     *
     * @param ms
     */
    private void sleep(long ms) {
        try {
            // 需要注意的是,这里休眠只是为了不让CPU飙百，避免没有数据来的情况下。
            // 当然这也意味着消费会受影响,每秒200条.
            Thread.sleep(ms);
        } catch (InterruptedException e) {
            e.printStackTrace();
            logger.error("数据传输线程失败", e);
        }
    }

    /**
     * 加速消费
     *
     * @param rate 消费阀值
     * @throws Exception
     */
    private void speedConsumer(int rate) throws Exception {
        for (int i = 0; i < rate; i++) {
            BaseDTO baseDTO = queueData.poll();
            sendOnWay(baseDTO);
        }
    }
}
