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();
        }

    }

}