import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.ArrayBlockingQueue;

import java.util.concurrent.locks.Condition;

import java.util.concurrent.locks.Lock;

import java.util.concurrent.locks.ReentrantLock;

/**

 * 实现可调整大小的阻塞队列，支持数据迁移平衡reader,writer读取吸入速度，达到最大吞吐

 * @author hanjie

 *

 */

public class RecordBuffer {

    public static final Record CLOSE_RECORD = new Record() {

        @Override

        public Object getColumnValue(String columnName) {

            // TODO Auto-generated method stub

            return null;

        }

    };

    public static final Record SWITCH_QUEUE_RECORD = new Record() {

        @Override

        public Object getColumnValue(String columnName) {

            // TODO Auto-generated method stub

            return null;

        }

    };

    public Lock switchingQueueLock = new ReentrantLock();

    public Condition readerSwitched = switchingQueueLock.newCondition();

    public Condition writerSwitched = switchingQueueLock.newCondition();

    public Condition switchFinished = switchingQueueLock.newCondition();

    public volatile boolean readerSwitchSuccess = true;

    public volatile boolean writerSwitchSuccess = true;

    public volatile boolean switchingQueue = false;

    public volatile boolean closed = false;

    private volatile ArrayBlockingQueue<Record> queue;

    private TaskCounter taskCounter;

    public RecordBuffer(TaskCounter taskCounter, int size) {

        this.queue = new ArrayBlockingQueue<Record>(size);

        this.taskCounter = taskCounter;

    }

    public void resize(int newSize) {

        try {

            if(closed){

                return;

            }

            switchingQueueLock.lock();

            try {

                //double check下，要不可能writer收到CLOSED_record已经 退出了。writerSwitched.await() 会hang住

                if(closed){

                    return;

                }

                this.switchingQueue = true;

                ArrayBlockingQueue<Record> oldQueue = queue;

                queue = new ArrayBlockingQueue<Record>(newSize);

                this.readerSwitchSuccess = false;

                this.writerSwitchSuccess = false;

                //先拯救下writer，可能writer刚好阻塞到take上,失败也没关系，说明老队列不空，writer不会阻塞到take

                oldQueue.offer(SWITCH_QUEUE_RECORD);

                while (!writerSwitchSuccess) {

                    writerSwitched.await();

                }

                //writer先切换队列，然后reader可能阻塞在最后一个put上,清空下老队列拯救reader，让它顺利醒来

                transferOldQueueRecordsToNewQueue(oldQueue);

                while (!readerSwitchSuccess) {

                    readerSwitched.await();

                }

                //前面的清空，刚好碰到reader要put最后一个，非阻塞式清空动作就有残留最后一个put

                transferOldQueueRecordsToNewQueue(oldQueue);

                this.switchingQueue = false;

                this.switchFinished.signalAll();

            } finally {

                switchingQueueLock.unlock();

            }

        } catch (InterruptedException e) {

            Thread.currentThread().interrupt();

            throw new RuntimeException(e);

        }

    }

    private void transferOldQueueRecordsToNewQueue(ArrayBlockingQueue<Record> oldQueue)

            throws InterruptedException {

        List<Record> oldRecords = new ArrayList<Record>(oldQueue.size());

        Record record = null;

        while ((record = oldQueue.poll()) != null) {

            oldRecords.add(record);

        }

        // 转移老队列剩下的记录到新队列

        for (int i = 0; i < oldRecords.size(); i++) {

            queue.put(oldRecords.get(i));

        }

    }

    public void close() {

        this.closed = true;

        switchingQueueLock.lock();

        try {

            //如果正在切换队列， 等切换做完才能，发送最后一个CLOSE

            while (switchingQueue) {

                switchFinished.await();

            }

            this.queue.put(CLOSE_RECORD);

        } catch (InterruptedException e) {

            Thread.currentThread().interrupt();

            throw new RuntimeException(e);

        }

        finally{

            switchingQueueLock.unlock();

        }

    }

    public void put(Record record) {

        try {

            if (!queue.offer(record)) {

                taskCounter.incrBufferFullCount();

                if (!readerSwitchSuccess) {

                    notifyReaderSwitchSuccess();

                }

                queue.put(record);

            }

            taskCounter.incrReadCount();

        } catch (InterruptedException e) {

            Thread.currentThread().interrupt();

            throw new RuntimeException(e);

        }

    }

    private void notifyReaderSwitchSuccess() {

        System.out.println("reader switch");

        switchingQueueLock.lock();

        try {

            readerSwitchSuccess = true;

            readerSwitched.signalAll();

        } finally {

            switchingQueueLock.unlock();

        }

    }

    public Record take() {

        try {

            Record record = queue.poll();

            //如果拿到了切换记录，则切换队列重试

            if(record == SWITCH_QUEUE_RECORD){

                if (!writerSwitchSuccess) {

                    notifyWriterSwitchSuccess();

                }

                record = queue.poll();

            }

            if (record == null) {

                taskCounter.incrBufferEmptyCount();

                //调用take先检查是否正在切换,保证拿到新的队列

                if (!writerSwitchSuccess) {

                    notifyWriterSwitchSuccess();

                }

                record = queue.take();

                //如果很不幸刚好在take阻塞时候，切换，只能发送一个切换记录将其唤醒

                if(record == SWITCH_QUEUE_RECORD){

                    if (!writerSwitchSuccess) {

                        notifyWriterSwitchSuccess();

                    }

                    record = queue.take();

                }

            }

            if (record == CLOSE_RECORD) {

                if (!writerSwitchSuccess) {

                    notifyWriterSwitchSuccess();

                }

                return null;

            }

            taskCounter.incrWriteCount();

            return record;

        } catch (InterruptedException e) {

            Thread.currentThread().interrupt();

            throw new RuntimeException(e);

        }

    }

    private void notifyWriterSwitchSuccess() {

        System.out.println("writer switch");

        switchingQueueLock.lock();

        try {

            writerSwitchSuccess = true;

            writerSwitched.signalAll();

        } finally {

            switchingQueueLock.unlock();

        }

    }

}