package com.twjitm.core.utils.time.timer;

import com.twjitm.core.common.config.global.GlobalConstants;
import com.twjitm.core.spring.SpringServiceManager;
import com.twjitm.threads.thread.NettyThreadNameFactory;

import java.util.ArrayList;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * 服务器：轮询时间片
 * 定时轮是一种数据结构，其主体是一个循环列表（circular buffer），每个列表中包含一个称之为槽（slot）的结构（附图）。
 * 至于 slot 的具体结构依赖具体应用场景。
 * 以本文开头所述的管理大量连接 timeout 的场景为例，描述一下 timing wheel的具体实现细节。
 * 定时轮的工作原理可以类比于始终，如上图箭头（指针）按某一个方向按固定频率轮动，每一次跳动称为一个 tick。
 * 这样可以看出定时轮由个3个重要的属性参数，ticksPerWheel（一轮的tick数），tickDuration（一个tick的持续时间）
 * 以及 timeUnit（时间单位），
 * 例如 当ticksPerWheel=60，tickDuration=1，timeUnit=秒，这就和现实中的始终的秒针走动完全类似了。
 * <p>
 * 主要是用来销毁或者实现一些定时任务等
 *
 * @author twjitm - [Created on 2018-08-31 10:30]
 * @company https://github.com/twjitm/
 * @jdk java version "1.8.0_77"
 */
public class PollTimer<E> {
    /**
     * 持续检测时间
     */
    private long tickDuration;
    /**
     * 检测数量，时间槽的数量
     */
    private int ticksPerWheel;
    /**
     * 当前检测所在的位置索引
     */
    private volatile int currentTickIndex = 0;
    /**
     * 周期事物坚挺者
     */
    private CopyOnWriteArrayList<ExpirationListener<E>> expirationListeners =
            new CopyOnWriteArrayList();
    /**
     * 时间槽
     */
    private ArrayList<TimeSlot<E>> wheel;
    /**
     * 指示器
     */
    private Map<E, TimeSlot<E>> indicator = new ConcurrentHashMap();
    /**
     * 是否关闭标识
     */
    private AtomicBoolean shutdown = new AtomicBoolean(false);
    /**
     * 读写锁
     */
    private ReadWriteLock lock = new ReentrantReadWriteLock();
    /**
     * 轮询线程
     */
    private Thread workThread;

    /**
     * 构造一个轮询事物实体，用于执行周期性任务
     *
     * @param tickDuration
     * @param timeUnit
     * @param ticksPerWheel
     */
    public PollTimer(int tickDuration, TimeUnit timeUnit, int ticksPerWheel) {
        if (timeUnit == null) {
            throw new NullPointerException("轮询时间片单位为空");
        }
        if (tickDuration <= 0) {
            throw new IllegalArgumentException(" 持续时间必须大于0" + tickDuration);
        }

        this.wheel = new ArrayList<>();
        this.tickDuration = TimeUnit.MILLISECONDS.convert(tickDuration, timeUnit);
        this.ticksPerWheel = ticksPerWheel;
        for (int i = 0; i < this.ticksPerWheel; i++) {
            wheel.add(new TimeSlot<>(i));
        }

        wheel.trimToSize();
        /**
         * 通过ThreadFactory 方式来创建线程
         */

        NettyThreadNameFactory factory = new NettyThreadNameFactory(
                GlobalConstants.Thread.POLL_LIFE_CYCLE);
        workThread = factory.newThread(new TickWorker());
    }

    /**
     * 开启周期轮询
     */
    public void start() {
        if (shutdown.get()) {
            throw new IllegalStateException(GlobalConstants.Thread.POLL_LIFE_CYCLE + "线程已经被停止");
        }

        if (!workThread.isAlive()) {
            workThread.start();
        }
    }

    /**
     * 定制周期轮询
     *
     * @return
     */
    public boolean stop() {
        if (!shutdown.compareAndSet(false, true)) {
            return false;
        }

        boolean interrupted = false;
        //发出中断信号
        while (workThread.isAlive()) {
            workThread.interrupt();
            try {
                //等待线程执行完毕。将线程结束，抛出异常
                workThread.join(1000);
            } catch (Exception e) {
                interrupted = true;
            }
        }

        if (interrupted) {
            Thread.currentThread().interrupt();
        }

        return true;
    }

    /**
     * 添加一个周期监听器
     *
     * @param listener
     */
    public void addExpirationListener(ExpirationListener<E> listener) {
        expirationListeners.add(listener);
    }

    /**
     * 移除一个周期监听器
     *
     * @param listener
     */
    public void removeExpirationListner(ExpirationListener<E> listener) {
        expirationListeners.remove(listener);
    }

    /**
     * 增加一个时间片
     *
     * @param e
     * @return
     */
    public long add(E e) {
        synchronized (e) {
            checkAdd(e);
            int previousTickIndex = getPreviousTickIndex();
            TimeSlot<E> timeSlotSet = wheel.get(previousTickIndex);
            timeSlotSet.add(e);
            indicator.put(e, timeSlotSet);
            return (ticksPerWheel - 1) * tickDuration;
        }
    }

    public boolean remove(E e) {
        synchronized (e) {
            TimeSlot<E> timeSlot = indicator.get(e);
            if (timeSlot == null) {
                return false;
            }

            indicator.remove(e);
            return timeSlot.remove(e);
        }
    }

    /**
     * 执行周期任务后需要通知监听器。
     *
     * @param idx
     */
    public void notifyExpired(int idx) {


        TimeSlot<E> timeSlot = wheel.get(idx);
        Set<E> elements = timeSlot.getElements();
        for (E e : elements) {
            timeSlot.remove(e);
            synchronized (e) {
                TimeSlot<E> latestSLot = indicator.get(e);
                if (latestSLot.equals(timeSlot)) {
                    indicator.remove(e);
                }
            }

            for (ExpirationListener<E> listener : expirationListeners) {
                listener.expired(e);
            }
        }
    }

    /**
     * 获取当前tick的前一个检点
     *
     * @return
     */
    private int getPreviousTickIndex() {
        lock.readLock().lock();
        try {
            int cti = currentTickIndex;
            if (cti == 0) {
                return ticksPerWheel - 1;
            }
            return cti - 1;
        } catch (Exception e) {

        } finally {
            lock.readLock().unlock();
        }

        return currentTickIndex - 1;
    }

    private void checkAdd(E e) {
        TimeSlot<E> slot = indicator.get(e);
        if (slot != null) {
            slot.remove(e);
        }
    }


    /**
     * 周期任务
     */
    private class TickWorker implements Runnable {

        /**
         * 启动时间
         */
        private long startTime;

        /**
         * 运行次数
         */
        private long tick = 1L;


        @Override
        public void run() {
            startTime = System.currentTimeMillis();
            tick = 1;
            for (int i = 0; !shutdown.get(); i++) {
                if (i == wheel.size()) {
                    i = 0;
                }

                lock.writeLock().lock();
                try {
                    currentTickIndex = i;
                } catch (Exception e) {

                } finally {
                    lock.writeLock().unlock();
                }

                notifyExpired(currentTickIndex);
                waitForNexTick();
            }
        }

        /**
         * 等待一个时间片
         */
        private void waitForNexTick() {
            for (; ; ) {
                long currentTime = System.currentTimeMillis();
                long sleepTime = tickDuration * tick - (currentTime - startTime);
                if (sleepTime <= 0) {
                    break;
                }

                try {
                    Thread.sleep(sleepTime);
                } catch (Exception e) {

                } finally {
                    break;
                }
            }
        }
    }
}
