package com.twjitm.core.common.service.rpc.client;

import com.twjitm.core.common.config.global.NettyGameServiceConfigService;
import com.twjitm.core.common.netstack.entity.rpc.NettyRpcRequestMessage;
import com.twjitm.core.common.netstack.entity.rpc.NettyRpcResponseMessage;
import com.twjitm.core.common.service.rpc.service.NettyRpcProxyService;
import com.twjitm.core.spring.SpringServiceManager;
import com.twjitm.core.utils.logs.LoggerUtils;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 远程调用线程执行返结果对象，本身{@link NettyRpcFuture}继承
 * 了{@link Future}类，所以执行线程必定会阻塞，所以不能再
 * 游戏线程使用本对象，即不能再{@link io.netty.channel.EventLoop}所在的线程执行
 *
 * @author twjitm - [Created on 2018-08-20 11:38]
 * @company https://github.com/twjitm/
 * @jdk java version "1.8.0_77"
 */
public class NettyRpcFuture implements Future<Object> {
    private Logger logger = LoggerUtils.getLogger(NettyRpcFuture.class);
    /**
     * 同步器
     */
    private Sync sync;
    /**
     * rpc 请求消息
     */
    private NettyRpcRequestMessage request;
    /**
     * rpc返回消息
     */
    private NettyRpcResponseMessage response;
    /**
     * 开始时间
     */
    private long startTime;

    /**
     * 回调接口
     */
    private List<NettyAsyncRPCCallback> pendingCallbacks = new ArrayList<NettyAsyncRPCCallback>();
       /**
     * 结果检测锁
     */
    private ReentrantLock lock = new ReentrantLock();

    public NettyRpcFuture(NettyRpcRequestMessage request) {
        this.sync = new Sync();
        this.request = request;
        this.startTime = System.currentTimeMillis();
    }

    /**
     * 是否完成 利用同步锁状态{@link AbstractQueuedSynchronizer}
     * 来检测是否执行完成。
     *
     * @return
     */
    @Override
    public boolean isDone() {
        return sync.isDone();
    }

    /**
     * 获取一个返回结果
     *
     * @return
     */
    @Override
    public Object get() {
        //阻塞等待，一直等到返回结果到来，处于阻塞状态
        sync.acquire(-1);
        if (this.response != null) {
            return this.response.getResult();
        } else {
            return null;
        }
    }

    /**
     * 获取一个是否具有过时效果的返回结果
     *
     * @param timeout
     * @param unit
     * @return
     * @throws InterruptedException
     */
    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException {
        //阻塞到获取锁返回true表示获得了锁，或者被打断抛出异常，或者到超时，返回false表示没有获得锁。
        boolean success = sync.tryAcquireNanos(-1, unit.toNanos(timeout));
        if (success) {
            //获得锁成功，表明已经有结果返回或者还没有提交处理，中断处理了
            if (this.response != null) {
                return this.response.getResult();
            } else {
                return null;
            }
        } else {
            throw new RuntimeException("TIMEOUT EXCEPTION. REQUEST ID: " + this.request.getRequestId()
                    + ". REQUEST CLASS NAME: " + this.request.getClassName()
                    + ". REQUEST METHOD: " + this.request.getMethodName());
        }
    }

    /**
     * 是否能够中断一个rpc请求消息
     *
     * @return
     */
    @Override
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }

    /**
     * 中断一个rpc请求消息
     *
     * @param mayInterruptIfRunning 中断的时候是否正在有运行的任务
     * @return
     */
    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }

    /**
     * 收到rpc消息返回。
     * 收到一个rpc消息返回的时候，首先将消息保存到本地，然后将同步锁释放掉。{@link Sync#release(int)}
     *
     * @param response
     */
    public void done(NettyRpcResponseMessage response) {
        this.response = response;
        sync.release(1);
        invokeCallbacks();
        // Threshold
        long responseTime = System.currentTimeMillis() - startTime;
        /**
         * 远程调用返回接口最大时长
         */
        long responseTimeThreshold = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService().getNettyGameServiceConfig().getRpcTimeOut();
        if (responseTime > responseTimeThreshold) {
            logger.warn("SERVICE RESPONSE TIME IS TOO SLOW. REQUEST ID = " + response.getRequestId() + ". RESPONSE TIME = " + responseTime + "ms");
        }
    }

    /**
     * 是否超时，当网络状态比较差或者负载比较高的时候，一条rpc请求消息可能会延迟，
     * 可以利用延迟策略来决定消息是否重新发送处理。
     *
     * @return
     */
    public boolean isTimeout() {
        long responseTime = System.currentTimeMillis() - startTime;
        NettyGameServiceConfigService gameServerConfigService = SpringServiceManager.getSpringLoadService().getNettyGameServiceConfigService();
        int timeOut = gameServerConfigService.getNettyGameServiceConfig().getRpcFutureDeleteTimeOut();
        if (responseTime >= timeOut) {
            return true;
        }
        return false;
    }

    /**
     * 调用回调函数。当一个rpc 请求消息有多个回调函数调用的时候
     * 需要把回调函数接口存放到一个集合中，才用可重入锁{@link ReentrantLock}
     * 来解决并发带来的问题
     */
    private void invokeCallbacks() {
        lock.lock();
        try {
            for (final NettyAsyncRPCCallback callback : pendingCallbacks) {
                runCallback(callback);
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * 添加回调函数，添加回调函数即对当前rpc请求返回结果进行
     * 监听处理
     *
     * @param callback
     * @return
     */
    public NettyRpcFuture addCallback(NettyAsyncRPCCallback callback) {
        lock.lock();
        try {
            if (isDone()) {
                logger.info("远程调用结果已经获取到了。直接回调一个函数");
                runCallback(callback);
            } else {
                logger.info("等待远程执行结果的到来，需要将回调函数放入到队列中");
                this.pendingCallbacks.add(callback);
            }
            //不管怎么样，都要释放锁
        } finally {
            lock.unlock();
        }
        return this;
    }

    /**
     * 运行一个回调。如何获得一个执行结果呢？需要在{@link NettyRpcProxyService#submit(Runnable)}
     * 的回调函数里面获取返回结果。最后将结果类型是否成功提交给调用线程的{@link NettyAsyncRPCCallback}对象，
     * {@link NettyAsyncRPCCallback} 对象自己去实现返回成功的业务逻辑和返回失败的业务逻辑
     *
     * @param callback
     */
    private void runCallback(final NettyAsyncRPCCallback callback) {
        final NettyRpcResponseMessage res = this.response;
        NettyRpcProxyService nettyRpcProxyService = SpringServiceManager.getSpringLoadService().getNettyRpcProxyService();
        nettyRpcProxyService.submit(() -> {
            if (!res.isError()) {
                callback.success(res.getResult());
            } else {
                callback.fail(new RuntimeException("RESPONSE ERROR", new Throwable(res.getError())));
            }
        });
    }

    /**
     * 实现异步回调的关键核心
     */
    static class Sync extends AbstractQueuedSynchronizer {

        private static final long serialVersionUID = 1L;

        private final int done = 1;
        private final int pending = 0;

        @Override
        protected boolean tryAcquire(int acquires) {
            return getState() == done;
        }

        /**
         * CAS操作，保证原子性
         *
         * @param releases
         * @return
         */
        @Override
        protected boolean tryRelease(int releases) {
            if (getState() == pending) {
                if (compareAndSetState(pending, done)) {
                    return true;
                }
            }
            return false;
        }

        public boolean isDone() {
            getState();
            return getState() == done;
        }
    }
}
