package org.kne.cloud.network.congestion;

import java.io.Closeable;
import java.io.IOException;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.UUID;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.function.Consumer;

import org.kne.cloud.network.NetworkPacket;
import org.kne.cloud.network.ThreadTool;
import org.kne.cloud.network.ipv6.IPv6Packet;
import org.kne.cloud.network.klalb.DATATPacket;
import org.kne.cloud.network.klalb.SendItem;
import org.kne.cloud.network.kltp.KLTPPacket;
import org.kne.concurrent.ThreadParker;

public class ReceivePacketSlidingWindow<K, V extends NetworkPacket>  implements Closeable, AutoCloseable{
	private ThreadParker tp=new ThreadParker();
	
	private Map<K,SendItem<V>> recvMap = new ConcurrentHashMap<K,SendItem<V>>();
	private AtomicLong recvWindowUsed=new AtomicLong(0);
	//private LongAdder recvWindowUsed=new LongAdder();
	private long recvWindowSize;
	private int headerCalibrate = 0;

	private volatile boolean closed = false;
	
	
	public ReceivePacketSlidingWindow(long recvWindowSize) {
		super();
		this.recvWindowSize = recvWindowSize;
	}
	
	public ReceivePacketSlidingWindow(long recvWindowSize, int headerCalibrate) {
		super();
		this.recvWindowSize = recvWindowSize;
		this.headerCalibrate = headerCalibrate;
	}

	public void setHeaderCalibrate(int headerCalibrate) {
		this.headerCalibrate = headerCalibrate;
	}
	
	public int getHeaderCalibrate() {
		return headerCalibrate;
	}
	
	public void put(K sequence, V packet)  {
		SendItem<V> newitem = new SendItem<V>(packet,headerCalibrate);
		SendItem<V> old = recvMap.putIfAbsent(sequence, newitem);
		long dx=0;
		if (old != null) { 
			dx=-old.getPacketLength() ;
		}
		recvWindowUsed.addAndGet(newitem.getPacketLength() +dx);
		tp.unpark();
	}
	
	public V poll(K key) {
		SendItem<V> dtp2 = recvMap.remove(key);
		if (dtp2 != null) {
			recvWindowUsed.addAndGet(-dtp2.getPacketLength());
			return dtp2.getPacket();
		}else {
			return null;
		}
	}
	public V take(K key) throws SocketException {
		while(true) {
			V val=poll(key);
			if(val!=null) {
				return val;
			}
			if(isClosed()) {
				throw new SocketException("Receive window closed!");
			}tp.parkNanos(1000000L);
		}
	}
	
	public V take(K key,long timeoutNanos) throws SocketException, SocketTimeoutException {
		long start=System.nanoTime();
		while(true) {
			if(System.nanoTime()-start>timeoutNanos) {
				throw new SocketTimeoutException("Receive time out:"+(System.nanoTime()-start)+">"+timeoutNanos);
			}
			V val=poll(key);
			if(val!=null) {
				return val;
			}
			if(isClosed()) {
				throw new SocketException("Receive window closed!");
			}tp.parkNanos(1000000L);
		}
	}
	
	public long getRecvWindowUsed() {
		return recvWindowUsed.get();
	}

	public long getRecvWindowAvaliable() {
		return Math.max(0, recvWindowSize-recvWindowUsed.get());
	}
	
	public Map<K, SendItem<V>> getRecvMap() {
		return recvMap;
	}

	public int getWindowPacketCount() {
		return recvMap.size();
	}

	public boolean isEmpty() {
		return recvMap.isEmpty();
	}
	
	public boolean isClosed() {
		return closed;
	}

	@Override
	public void close() {
		closed = true;
	}

	@Override
	public String toString() {
		return "ReceivePacketSlidingWindow [recvWindowUsed=" + recvWindowUsed + ", recvWindowSize=" + recvWindowSize
				+ ", closed=" + closed + "]";
	}

	

	
}