package org.kne.cloud.network.congestion;
import java.lang.ref.Cleaner;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.LockSupport;
import java.util.function.Consumer;

import org.kne.cloud.network.ThreadTool;
import org.kne.concurrent.SpinLock;
import org.kne.opencl64.Releaser;

/**
 * 网络消息批量发送器，用于减少小包数量，降低网络性能开销
 * @param <T> 消息类型
 */

public class ArrayListMessageBatcher<T> implements MessageBatcher<T>{
	private static final Cleaner clr=Cleaner.create();
	
	private ArrayListMessageBatcher0<T> impl;
	
	private ArrayListMessageBatcherReleaser<T> releaser;
	
	public ArrayListMessageBatcher(int batchSize, long maxDelay) {
		impl=new ArrayListMessageBatcher0<T>(batchSize, maxDelay);
		releaser=new ArrayListMessageBatcherReleaser<T>(impl);
		clr.register(this, releaser);
	}

	@Override
	public int getBatchSize() {
		return impl.getBatchSize();
	}

	@Override
	public long getMaxDelay() {
		return impl.getMaxDelay();
	}

	@Override
	public void setConsumer(Consumer<List<T>> consumer) {
		impl.setConsumer(consumer);
	}

	@Override
	public void putMessage(T message) {
		impl.putMessage(message);
	}

	@Override
	public void putMessages(List<T> messages) {
		impl.putMessages(messages);
	}

	@Override
	public void flush() {
		impl.flush();
	}

	@Override
	public int getQueueSize() {
		return impl.getQueueSize();
	}

	@Override
	public boolean isClosed() {
		return impl.isClosed();
	}

	@Override
	public void close() {
		impl.close();
	}
	public static void main(String[] args) throws InterruptedException {
		/*for(;;){
			ArrayListMessageBatcher<Integer>mes=new ArrayListMessageBatcher<>(10, 1000000L);
			System.gc();
		}*/
		ArrayListMessageBatcher<Integer>mes=new ArrayListMessageBatcher<>(10, 1000000L);
		mes.setConsumer((x)->{System.out.println(x);});
		int n=0;
		for(;;){
			mes.putMessage(n++);
			mes.putMessage(n++);
			mes.putMessage(n++);
			mes.putMessage(n++);
			Thread.sleep(1);
		}
	}
}
class ArrayListMessageBatcherReleaser<T> extends Releaser<ArrayListMessageBatcher0<T>>{

	public ArrayListMessageBatcherReleaser(ArrayListMessageBatcher0<T> resource) {
		super(resource);
	}

	@Override
	protected void release(ArrayListMessageBatcher0<T> resource) {
		resource.close();
		//System.out.println("release!");
	}
	
}
 class ArrayListMessageBatcher0<T> implements Runnable,MessageBatcher<T>{
    private ArrayList<T> messageList;
    private Lock lock=new SpinLock();
    private final int batchSize;
    private final long maxDelay;
    private Consumer<List<T>> consumer;
    private final Thread batchThread;
    private final AtomicBoolean closed = new AtomicBoolean(false);
	private long firstTime;
    
    /**
     * 构造函数
     * @param batchSize 批量大小，当消息达到此数量时触发发送
     * @param maxDelay 最大延迟时间(毫秒)，即使消息数量不足也会触发发送
     */
    public ArrayListMessageBatcher0(int batchSize, long maxDelay) {
        this.batchSize = batchSize;
        this.maxDelay = maxDelay;
        
        // 创建并启动批量处理线程
        this.batchThread = ThreadTool.makeVDaemonThread("ArrayListMessageBatcher Thread", this );
        this.batchThread.setDaemon(true);
        this.batchThread.start();
    }
    
    /**
     * 设置消息消费者回调函数
     * @param consumer 消费者回调
     */
    public void setConsumer(Consumer<List<T>> consumer) {
        this.consumer = consumer;
    }
    
    /**
     * 添加消息到批量处理器
     * @param message 消息对象
     */
    public void putMessage(T message) {
        if (message != null) {
        	lock.lock();
        	try {
        		if(messageList==null) {
        			messageList=new ArrayList<T>(batchSize);
        			firstTime=System.nanoTime();
        		}
        	messageList.add(message);
        	check(false);
        	}finally {
        		lock.unlock();
        	}
        }
    }

    
    /**
     * 批量添加消息
     * @param messages 消息列表
     */
    public void putMessages(List<T> messages) {
    	for(T message :messages) {
    		putMessage(message);
    	}
    }
    
    
    private void check(boolean force) {
    	if(messageList==null)
    		return;
    	long currTime=System.nanoTime();
		if(messageList.size()>=batchSize||(currTime-firstTime)>maxDelay||force) {
			if(consumer!=null) {
				try {
				consumer.accept(messageList);
				}catch(Exception e) {
					e.printStackTrace();
				}
			}
			messageList=null;
		}
	}

	/**
     * 停止批量处理器
     */
    public void close() {
        closed.set(true);
        LockSupport.unpark(batchThread); 
        // 确保所有消息都被发送
        flush();
    }
    
    /**
     * 批量处理消息的核心方法
     */
    public void run() {
    	while(!closed.get()) {
    		lock.lock();
        	try {
        		check(false);
        	}finally {
        		lock.unlock();
        	}
        	// 更精确的等待策略
            long sleepTimeNanos = calculateSleepTime();
            if (sleepTimeNanos > 0) {
                LockSupport.parkNanos(sleepTimeNanos);
            } else {
                // 避免忙等待
                LockSupport.parkNanos(1_000_000L); // 1ms
            }
    	}
    }
    
    /**
     * 计算需要等待的时间
     * @return 等待时间(纳秒)
     */
    private long calculateSleepTime() {
        lock.lock();
        try {
            if (messageList == null || messageList.isEmpty()) {
                return 10_000_000L; // 10ms
            }
            
            long elapsed = System.nanoTime() - firstTime;
            long remaining = maxDelay - elapsed;
            
            if (remaining <= 0) {
                return 0; // 立即处理
            }
            
            // 返回剩余时间或10ms中的较小值
            return Math.min(remaining, 10_000_000L);
        } finally {
            lock.unlock();
        }
    }
    
    public int getBatchSize() {
		return batchSize;
	}

	public long getMaxDelay() {
		return maxDelay;
	}

	public void flush() {
		lock.lock();
    	try {
    		check(true);
    	}finally {
    		lock.unlock();
    	}
	}
	
	public int getQueueSize() {
    	List<T> cmessageList =messageList;
    	if(cmessageList==null)
    		return 0;
        return cmessageList.size();
    }
	
	@Override
	public boolean isClosed() {
		return closed.get();
	}
	

	
}