package org.kne.cloud.network.congress;

import java.io.Closeable;
import java.io.IOException;
import java.net.SocketException;
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.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.ipv6.IPv6Packet;
import org.kne.cloud.network.klalb.SendItem;

public class SendPacketSlidingWindow <K,V extends NetworkPacket> implements Closeable,AutoCloseable{
	private Map<K,SendItem<V>> sendMap= new ConcurrentHashMap<>(1024,0.2f);
	//private AtomicInteger sendmapWindowUsed=new AtomicInteger(0);
	private LongAdder sendmapWindowUsed=new LongAdder();
	
	private int headerCalibrate=0;
	public void setHeaderCalibrate(int headerCalibrate) {
		this.headerCalibrate = headerCalibrate;
	}

	private CongressAlgorithm algorithm;
	
	private Consumer<V> resendConsumer;
	private Consumer<V> closingConsumer;
	private volatile boolean closed=false;
	private ResendChecker checker=new ResendChecker();
	private AtomicLong windowSize = new AtomicLong();
	

	private Lock lock;

	private Condition condition;
	
	public SendPacketSlidingWindow(CongressAlgorithm algorithm, long windowSize) {
		super();
		this.algorithm = algorithm;
		this.windowSize .set( windowSize);
		Thread t=new Thread(checker);
		t.start();
	}

	public SendPacketSlidingWindow(CongressAlgorithm algorithm, long windowSize,int headerCalibrate) {
		this(algorithm,windowSize);
		this.headerCalibrate=headerCalibrate;
	}
	
	public long getWindowSize() {
		return windowSize.get();
	}

	public void setWindowSize(long windowSize) {
		this.windowSize.set(windowSize);
	}

	public Consumer<V> getResendConsumer() {
		return resendConsumer;
	}

	public void setResendConsumer(Consumer<V> resendConsumer) {
		this.resendConsumer = resendConsumer;
	}

	public Consumer<V> getClosingConsumer() {
		return closingConsumer;
	}

	public void setClosingConsumer(Consumer<V> closingConsumer) {
		this.closingConsumer = closingConsumer;
	}

	public long getSendmapWindowUsed() {
		return sendmapWindowUsed.sum();
	}

	public int getHeaderCalibrate() {
		return headerCalibrate;
	}

	public CongressAlgorithm getAlgorithm() {
		return algorithm;
	}

	
	public boolean isCongress(IPv6Packet iPv6Packet,double scale) {
		boolean congress=sendmapWindowUsed.sum()>windowSize.get()*scale;
		return congress;	
		
	}
	
	public void put(K sequence,V packet) throws SocketException {
		if(closed) {
			throw new SocketException("sliding window closed!");
		}
		SendItem<V> newitem=new SendItem<V>( packet);
		SendItem<V> old= sendMap.put(sequence,newitem);
		if(old!=null) {
			sendmapWindowUsed.add( -(old.getPacketLength()+headerCalibrate));
		}
		sendmapWindowUsed.add( (newitem.getPacketLength()+headerCalibrate));
		long newWindow=sendmapWindowUsed.sum();
		algorithm.setCurrentWindowUsed(newWindow);
	}
	
	public V ack(K sequence) {
		SendItem<V> ipv6;
		if((ipv6=sendMap.remove(sequence))!=null) {
			sendmapWindowUsed.add( -(ipv6.getPacketLength()+headerCalibrate));
			long newWindow= sendmapWindowUsed.sum();
			algorithm.setCurrentWindowUsed(newWindow);
			long RTTC=System.nanoTime()-ipv6.getSendtime();
			algorithm.putAck(ipv6.getPacketLength(), RTTC, false);
			//System.out.println("remove:"+aseq.getSequence()+"  "+sendMap.size());
			Lock lockx=lock;
			if(lockx!=null) {
				lockx.lock();
				try {
					Condition cds=condition;
					if(cds!=null) {
						cds.signalAll();
					}
				}finally {
					lockx.unlock();
				}
			}
			return ipv6.getPacket();
		}else {
		//	System.out.println("miss:"+aseq.getSequence()+"  "+sendMap.size());
			return null;
		}
	}
	
	private class ResendChecker implements Runnable{ 

		@Override
			public void run() {
			while(!closed) {

				Collection<SendItem<V>>vals= sendMap.values();
				for (Iterator<SendItem<V>> iterator = vals.iterator(); iterator.hasNext();) {
					SendItem<V> sitm = (SendItem<V>) iterator.next();
					if(System.nanoTime() -sitm.getSendtime()>algorithm.getRTO()) {
						iterator.remove();
						 sendmapWindowUsed.add((int) -(sitm.getPacketLength()+headerCalibrate));
						 long newWindow=sendmapWindowUsed.sum();
						algorithm.setCurrentWindowUsed(newWindow);
						//if(!kplink.isStream())
						V pktr=sitm.getPacket();
						pktr.setPriority(pktr.getPriority()-1);
						if(resendConsumer!=null)
					resendConsumer.accept(pktr); 
					//System.out.println("reroute");
					}
				}
				
				
				
				try {
					Thread.sleep(2);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			// 清理资源
            if (closingConsumer != null) {
                for (SendItem<V> item : sendMap.values()) {
                    closingConsumer.accept(item.getPacket());
                }
            }
			sendMap.clear();
			sendmapWindowUsed.reset();
			}
	}
	public boolean isClosed() {
		return closed;
	}

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

	public void setCongressCondition(Lock lock2, Condition condition2) {
		this.lock=lock2;
				this.condition=condition2;
	}

	public Map<K, SendItem<V>> getSendmap() {
		return sendMap;
	}
	
}