package com.alibaba.otter.canal.sink.entry.group;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import com.alibaba.otter.canal.store.model.Event;

/**
 * 时间归并控制
 * 
 * <pre>
 * 大致设计：
 *  1. 多个队列都提交一个timestamp，判断出最小的一个timestamp做为通过的条件，然后唤醒<=该最小时间的线程通过
 *  2. 只有当多个队列都提交了一个timestamp，缺少任何一个提交，都会阻塞其他队列通过。(解决当一个库启动过慢或者发生主备切换时出现延迟等问题)
 * 
 * 存在一个假定，认为提交的timestamp是一个顺序递增，但是在两种case下会出现时间回退
 * a. 大事务时，事务头的时间会晚于事务当中数据的时间，相当于出现一个时间回退
 * b. 出现主备切换，从备机上发过来的数据会回退几秒钟
 * 
 * </pre>
 * 
 * @author jianghang 2012-10-15 下午10:01:53
 * @version 1.0.0
 */
public class TimelineBarrier implements GroupBarrier<Event> {

    protected int                 groupSize;
    protected ReentrantLock       lock           = new ReentrantLock();
    protected Condition           condition      = lock.newCondition();
    protected volatile long       threshold;
    protected BlockingQueue<Long> lastTimestamps = new PriorityBlockingQueue<Long>(); // 当前通道最后一次single的时间戳

    public TimelineBarrier(int groupSize){
        this.groupSize = groupSize;
        threshold = Long.MIN_VALUE;
    }

    /**
     * 判断自己的timestamp是否可以通过
     * 
     * @throws InterruptedException
     */
    public void await(Event event) throws InterruptedException {
        long timestamp = getTimestamp(event);
        try {
            lock.lockInterruptibly();
            single(timestamp);
            while (isPermit(event, timestamp) == false) {
                condition.await();
            }
        } finally {
            lock.unlock();
        }
    }

    /**
     * 判断自己的timestamp是否可以通过,带超时控制
     * 
     * @throws InterruptedException
     * @throws TimeoutException
     */
    public void await(Event event, long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
        long timestamp = getTimestamp(event);
        try {
            lock.lockInterruptibly();
            single(timestamp);
            while (isPermit(event, timestamp) == false) {
                condition.await(timeout, unit);
            }
        } finally {
            lock.unlock();
        }
    }

    public void clear(Event event) {
        // 出现中断有两种可能：
        // 1.出现主备切换，需要剔除到Timeline中的时间占位(这样合并时就会小于groupSize，不满足调度条件，直到主备切换完成后才能重新开启合并处理)
        // 2.出现关闭操作，退出即可
        lastTimestamps.remove(getTimestamp(event));
    }

    public void interrupt() {
        // do nothing，没有需要清理的上下文状态
    }

    public long state() {
        return threshold;
    }

    /**
     * 判断是否允许通过
     */
    protected boolean isPermit(Event event, long state) {
        return state <= state();
    }

    /**
     * 通知一下
     */
    protected void notify(long minTimestamp) {
        // 通知阻塞的线程恢复, 这里采用single all操作，当group中的几个时间都相同时，一次性触发通过多个
        condition.signalAll();
    }

    /**
     * 通知下一个minTimestamp数据出队列
     * 
     * @throws InterruptedException
     */
    private void single(long timestamp) throws InterruptedException {
        lastTimestamps.add(timestamp);

        if (timestamp < state()) {
            // 针对mysql事务中会出现时间跳跃
            // 例子：
            // 2012-08-08 16:24:26 事务头
            // 2012-08-08 16:24:24 变更记录
            // 2012-08-08 16:24:25 变更记录
            // 2012-08-08 16:24:26　事务尾

            // 针对这种case，一旦发现timestamp有回退的情况，直接更新threshold，强制阻塞其他的操作，等待最小数据优先处理完成
            threshold = timestamp; // 更新为最小值
        }

        if (lastTimestamps.size() >= groupSize) {// 判断队列是否需要触发
            // 触发下一个出队列的数据
            Long minTimestamp = this.lastTimestamps.peek();
            if (minTimestamp != null) {
                threshold = minTimestamp;
                notify(minTimestamp);
            }
        } else {
            threshold = Long.MIN_VALUE;// 如果不满足队列长度，需要阻塞等待
        }
    }

    private Long getTimestamp(Event event) {
        return event.getExecuteTime();
    }

}
