package pku;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

public class BufferQueue {
 //设置一个缓冲区大小为8k
 private static final int BUFF_SIZE = 4096*2;

 //并发写入线程池线程数
 private static final int N_THREAD = 20;

 //用于记录刷盘了几个块,使用此值乘上缓冲区大小就是每个块的起始地址
 private static AtomicLong wrotePosition = new AtomicLong();

 private volatile boolean firstGet = true;
 //JavaNIO对byte数组的一种封装
 //静态方法初始化
 private ByteBuffer buffer = ByteBuffer.allocateDirect(BUFF_SIZE);
 BufferWrapper bufferWrapper = new BufferWrapper(buffer);

 //记录当前队列已经有几个消息了
 private int queueLength = 0;

 //多线程刷盘的线程池,目前未使用,待优化
 //private static ThreadPoolExecutor flushThread = (ThreadPoolExecutor) Executors.newFixedThreadPool(N_THREAD);
 //

 //创建一个定长线程池,可控制线程最大并发数,超出的线程会在队列中等待。
 //通过Executors方法创建
 private static ThreadPoolExecutor flushThread =
 new ThreadPoolExecutor(N_THREAD, N_THREAD,0L,
 TimeUnit.MILLISECONDS,
 new LinkedBlockingQueue<Runnable>());
 //用于channel的管理,来源于rabbitmq
 private static ChannelManager channelManager;


 /* static块
 * static关键字还有一个比较关键的作用,
 * 用来形成静态代码块(static{}即static块)以优化程序性能。
 * static块可以置于类中的任何地方,类中可以有多个static块。
 * 在类初次被加载的时候执行且仅会被执行一次,
 * 会按照static块的顺序来执行每个static块,一般用来初始化静态变量和调用静态方法。
 * 初始化队列,根据设置的访问线程数创建channel池,每个channel连向同一个文件
 */
 static {
 //创建channel池,用后面的管理类去管理
 //所有的channel连向一个文件
 FileChannel[] channels = new FileChannel[N_THREAD];
 for (int i = 0; i < N_THREAD; i++) {
 try {
 RandomAccessFile randomAccessFile = new RandomAccessFile("data/queue_data", "rw");
 channels[i] = randomAccessFile.getChannel();
 } catch (FileNotFoundException e) {
 e.printStackTrace();
 }
 }
 channelManager = new ChannelManager(channels);
 }

 //一个list,list中的元素是TabelItem,记录每一个块信息(块起始物理地址,块内最后一个消息的序号)
 //我需要知道offset在第几页
 //存储页号,以及这一页存了多少消息
 //其实反过来就好理解了
 //本队列第N个消息,存储在文件的第几个BUFFER块中
 private BufferTable table = new BufferTable();


 public BufferQueue() {

 }

 /**
 * 将消息加入到消息队列缓冲区中,如果缓冲区能容纳消息,则消息直接加入缓冲区,
 * 若缓冲区容纳不下,则调用flush函数将整个buffer块写入文件
 * 写入大小即buffer块的大小
 * @param message 进入队列的消息
 */
 public void put(byte[] message) {
 //缓冲区落盘
 //2B记录消息长度,然后记录消息
 //一个buffer装不下了,将buffer刷盘,一个buffer自然成一块
 if (message.length + 2 > buffer.remaining()) {
 flush();
 }
 buffer.putShort((short) message.length);
 buffer.put(message);
 this.queueLength++;
 }

 /**
 * 刷盘操作,每次写到文件中的数据为一个缓冲区中所有数据(记为一块),
 * 首先将块信息记录到自定义的文件索引表中,然后从channel池中取出一个channel写入文件
 * 以一个buffer为单位写数据,buffer未必完全满
 */
 private void flush() {
 	//getAndIncrement函数为i++的原子操作,加锁,保证线程安全
 	/* 源代码
 	 * public final int getAndIncrement() {
		 *	 for (;;) {
			 int current = get();
			 int next = current + 1;
			 if (compareAndSet(current, next))
			 return current;
			 }
			}
 	 */
 long writePos = wrotePosition.getAndIncrement();

 //本队列第queueLength-1个消息,存储在文件的第 writePos 个BUFFER块的末尾
 //也就是这个buffer块中都是这个队列的消息,顺序存储
 table.add(writePos * BUFF_SIZE, queueLength - 1);

 //多线程写入,待优化
 // TODO: 待优化
// flushThread.submit(new Runnable() {
// @Override
// public void run() {

 //调用clear()方法:position将被设回0,limit设置成capacity,
 //Buffer被清空了,其实Buffer中的数据并未被清除
 //只是这些标记告诉我们可以从哪里开始往Buffer里写数据。
 buffer.clear();
 try {
 FileChannel channel = channelManager.getChannel();
 channel.write(buffer, writePos * BUFF_SIZE);
 channelManager.free(channel);
 } catch (IOException e) {
 e.printStackTrace();
 }
//
// }
// });


// flushThread.submit(()->{
// buffer.clear();
// try {
// FileChannel channel = channelManager.getChannel();
// channel.write(buffer, writePos * BUFF_SIZE);
// channelManager.free(channel);
// } catch (IOException e) {
// e.printStackTrace();
// }
// });


 //清空缓冲区
 buffer.clear();
 }

 /**
 * 从队列中取出固定个数消息放入集合,先根据消息的起始序号找到应该读取消息的位置,
 * 依次顺序取出消息直到消息数量达到num。如果取空队列后消息数不足num则也直接返回,
 * 若队列为空则返回一个空集合
 *
 * @param offset 需要获取消息的起始序号
 * @param num 需要获取消息的个数
 * @return 一个装有出队消息的集合
 */
 public synchronized Collection<byte[]> get(long offset, long num) {
//

 if (firstGet) {
 //TODO:为什么第一次读要把缓冲区写到文件里
 //因为可能读取的消息包括在缓冲区里的部分
 	//因为之前最后一次写之后没有清空尚未写满的buffer
 flush();
 firstGet = false;
 }

 List<byte[]> result = new ArrayList<>();
 //已经读了多少条
 long totalGet = 0;
 //正在读第几条消息
 long current = offset;

 for (int i = 0; i < num; i++) {
 //重点解释这一行
 TableItemWrapper itemWrapper = table.offsetToFilePos(current);
 //itemWrapper中获得了块的起始地址(字节)和块内偏移(第几个消息)
 if (itemWrapper == null) {
 if (result == null || result.size() == 0)
 return DefaultQueueStoreImpl.EMPTY;
 else
 return result;
 }
 buffer.clear();
 //读盘
 try {
 FileChannel channel = channelManager.getChannel();
 channel.read(buffer, itemWrapper.offset);
 channelManager.free(channel);
 } catch (IOException e) {
 e.printStackTrace();
 }
 buffer.clear();
 // TODO: 跳过偏移量
 if (i == 0) {
 bufferWrapper.skip(itemWrapper.offsetInBlock);
 }
 // TODO: 读取一块内的消息
 for (; current <= itemWrapper.lastMessageCount; current++) {
 if (totalGet == num) {
 return result;
 }
 byte[] message = bufferWrapper.giveMeAMessageThankYou();

 result.add(message);
 totalGet++;
 }
 buffer.clear();
 }
 return result;
 }

// private void waitWrite() {
// while (!flushThread.isTerminated()) {
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// }
// }

 /**
 * 自定义的文件索引数据结构,包含一个块信息的list
 */
 private static class BufferTable {
 /**
 *list中记录了每个块的信息
 */
 private List<TableItem> table = new ArrayList<>();

 /**
 * 通过接收刷盘时计算得到的块的起始物理地址,以及块中包含消息的最大序号,创建块信息并
 * 加入到索引中
 *
 * @param offset 块的起始物理地址
 * @param lastMessageCount 块中包含消息的最大序号
 */
 public synchronized void add(long offset, int lastMessageCount) {
 table.add(new TableItem(offset, lastMessageCount));
 }

 //给一条消息 我需要读哪一页?

 /**
 * 根据消息的序号,计算此消息在哪一个块中,在块内哪个位置,使用顺序查找
 * 依次扫描每一个块的信息,看块中最后一个消息的序号是否超过给定的消息
 * 序号,如果超过则说明消息在这个块中,返回块的信息,通过序号的差值确定
 * 块内偏移(单位为消息个数)
 *
 * @param offset 消息在队列中的序号
 * @return 该方法返回一个包装类,记录了属于一个消息的块位置,块内偏移,块最大消息序号
 */
 public TableItemWrapper offsetToFilePos(long offset) {
 //这里指offset所在的块的第一个消息的offset,单位是消息个数
 long blockStart = 0;

 //二分查找
 //处理边界条件
 TableItem firstTableItem = table.get(0);
 if(firstTableItem.lastMessageCount >= offset){
 return new TableItemWrapper((int) offset, firstTableItem.offset, firstTableItem.lastMessageCount);
 }
 int i = 0;
 int j = table.size()-1;
 //如果当前tableItem的lastMessageCount超过offset且前一个tableItem的lastMessageCount不超过offset,
 //则此条消息就在当前tableItem中
 while(i <= j){
 int mid = (i + j) / 2;
 TableItem curTableItem = table.get(mid);
 if(curTableItem.lastMessageCount < offset){ //消息在后面的item中
 i = mid + 1;
 }
 else{ //消息可能在当前item或者前面的item中
 TableItem preTableItem = table.get(mid - 1);
 if(preTableItem.lastMessageCount < offset){ //就在当前的item
 blockStart = preTableItem.lastMessageCount + 1;
 return new TableItemWrapper((int) (offset - blockStart), curTableItem.offset, curTableItem.lastMessageCount);
 }
 //消息在前面的item中
 j = mid - 1;
 }
 }

 // TODO:优化为二分查找
// for (int i = 0; i < table.size(); i++) {
// TableItem tableItem = table.get(i);
// if (tableItem.lastMessageCount < offset) {
// blockStart = tableItem.lastMessageCount + 1;
// } else {
// return new TableItemWrapper((int) (offset - blockStart), tableItem.offset, tableItem.lastMessageCount);
// }
// }
 return null;
 }
 }

 /**
 * 包装类,记录了属于一个消息的块位置,块内偏移,块最大消息序号
 */
 public static class TableItemWrapper {
 public int offsetInBlock;
 public long offset;
 public long lastMessageCount;

 /**
 * 构造函数,其中块内偏移通过BufferTable中的offsetToFilePos(int)计算得到
 *
 * @param offsetInBlock 块内偏移,表示消息是块中第几个消息
 * @param offset 块的起始物理地址,单位为字节
 * @param lastMessageCount 块中消息的最大序号
 */
 public TableItemWrapper(int offsetInBlock, long offset, long lastMessageCount) {
 this.offsetInBlock = offsetInBlock;
 this.offset = offset;
 this.lastMessageCount = lastMessageCount;
 }
 }

 /**
 * 这个类表示一个块,记录块信息,offset为物理地址(磁盘偏移),
 * lastMessageCount表示块内最后一个消息是队列中第几个消息
 */
 private static class TableItem {
 //磁盘偏移
 public long offset;
 //最后一条消息是队列中的第几条消息
 //为什么用最后一条消息
 //如果用第一条消息,就不知道
 public int lastMessageCount;

 /**
 * 构造函数,块的起始物理地址由刷盘时计算,计算方法为第几次刷盘(第几个块)*缓冲区大小
 *
 * @param offset 块的起始物理地址
 * @param firstMessageCount 块内最后一条消息的序号
 */
 public TableItem(long offset, int firstMessageCount) {
 this.offset = offset;
 this.lastMessageCount = firstMessageCount;
 }
 }

 //一个功能类,可以在块中进行两个操作

 /**
 * 一个功能类,封装了对块的两个操作,一个是取下一条消息,一个是跳过n条消息
 * 用于在定位到块位置以后从块中拿出具体的消息
 */
 private static class BufferWrapper {
 ByteBuffer buffer;

 public BufferWrapper(ByteBuffer buffer) {
 this.buffer = buffer;
 }

 /**
 * 在块中跳过n条消息,顺序扫描,先扫描一个short单位,获得本条消息长度
 * 然后position位置往后移动消息个长度,刚好到达下个消息的位置,循环n次
 *
 * @param n 跳过消息数
 */
 public void skip(int n) {
// System.out.print("skip:" + n + " buffer remain: " + buffer.remaining());
 for (int i = 0; i < n; i++) {
 short messageLengh = buffer.getShort();
 int oldPos = buffer.position();
// System.out.println("messageLenth = "+messageLengh+" oldPos = " + oldPos + " newPos = " + (oldPos + messageLengh));
 buffer.position(oldPos + messageLengh);
 }
 }

 /**
 * 取下一条消息,在当前位置扫描short单位,获得消息长度,读取响应长度的消息放入byte数组中,返回此数组
 *
 * @return 装有消息的byte数组
 */
 //取下一条消息
 public byte[] giveMeAMessageThankYou() {
 short messageLengh = buffer.getShort();
 byte[] message = new byte[messageLengh];
 buffer.get(message);
 return message;
 }
 }


 //Channel池的管理类

 /**
 * 自定义的channel池,用于处理文件的并发写入。提供了管理channel池的三个方法:
 * 寻找空闲channel、获得channel、释放channel
 */
 public static class ChannelManager {

 /**
 * 信号量,数量为channel池中channel数量,控制线程排队获得channel
 * Semaphore也是一个线程同步的辅助类,
 * 可以维护当前访问自身的线程个数,并提供了同步机制。
 */
 private Semaphore semaphore;
 //private ReentrantLock accessLock = new ReentrantLock();

 /**
 * 空闲标记,标记每个channel是否空闲
 */
 private HashMap<FileChannel, Boolean> channelState = new HashMap<>();

 /**
 * 构造函数,初始化channel池,设置信号量为channel数,并且将所有channel设置为空闲
 *
 * @param channels 将要使用的channel组
 */
 public ChannelManager(FileChannel[] channels) {
 	//一共有几个进程连接了文件
 semaphore = new Semaphore(channels.length);
 for (FileChannel channel : channels) {
 channelState.put(channel, false);
 }
 }

 /**
 * 获得一个空闲的channel,先尝试获取信号量,得到权限,寻找一个空闲的channel并返回
 * 申请文件通道资源,并占用线程
 * @return 返回一个空闲的channel
 */
 public FileChannel getChannel() {
 FileChannel target = null;
 try {
 	//获得线程许可
 semaphore.acquire();

 target = findFree();
// System.out.println("get channel:"+target);

 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 return target;
 }

 /**
 * 释放线程以及文件资源
 * 释放channel,将channel对应的空闲标记设置为<tt>false</tt>,并且释放信号量
 *
 * @param channel 传入使用完的channel进行释放
 */
 public synchronized void free(FileChannel channel) {
 channelState.put(channel, false);
// System.out.println("free channel:"+channel);
// System.out.print("free a channel ");

 semaphore.release();
 }

 /**
 * 返回一个空闲的channel,并置忙
 * 寻找空闲channel,遍历channel组,对每个channel查询空闲标记,找到第一个空闲的channel返回出去
 *
 * @return 返回值为一个空闲的channel
 */
 private synchronized FileChannel findFree() {
 for (FileChannel channel : channelState.keySet()) {
 if (!channelState.get(channel)) {
 channelState.put(channel, true);
// System.out.print("get a channel ");

 return channel;
 }
 }
 return null;
 }

// private void printFree() {
// int i = 0;
// for (FileChannel channel : channelState.keySet()) {
// if (!channelState.get(channel)) {
// i++;
// }
// }
// System.out.println(i);
// }
 }
}