package org.kne.cloud.network.klalb;

import java.io.IOException;
import java.net.Socket;
import java.util.*;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

import org.kne.cloud.network.mport.ServiceElement;

public class LocalTCPConnection extends TCPConnection {
	private static AtomicLong sng=new AtomicLong(1);
	
private long cuid;
private ServiceElement serviceElement;


private Set<KLALBBlock> inputcache = new TreeSet<>();
private List<KLALBBlock> outputcache = new Vector<>();

private volatile long inputcount = 1;
private volatile long outputcount = 1;
	public LocalTCPConnection(Socket s) throws IOException {
		super(s);
		cuid=sng.getAndIncrement();
	}
	public LocalTCPConnection(Socket s,long uid) throws IOException {
		super(s);
		cuid=uid;
	}
	public long getCuid() {
		return cuid;
	}
	public void setCuid(long cuid) {
		this.cuid = cuid;
	}
	public ServiceElement getServiceElement() {
		return serviceElement;
	}
	public void setServiceElement(ServiceElement serviceElement) {
		this.serviceElement = serviceElement;
	}
	
	
	public Set<KLALBBlock> getInputcache() {
 		return inputcache;
	}
	public List<KLALBBlock> getOutputcache() {
		return outputcache;
	}
	public long getInputcount() {
		return inputcount;
	}
	public long getOutputcount() {
		return outputcount;
	}
	public void unpackBlock(KLALBBlock data) throws IOException {
		if(data.number<inputcount) {
			return;
		}
		inputcache.add(data);
		for (Iterator<KLALBBlock> iterator = inputcache.iterator(); iterator.hasNext();) {
			KLALBBlock klalbBlock = (KLALBBlock) iterator.next();
			if(klalbBlock.number==inputcount) {
				iterator.remove();
				getDout().write(klalbBlock.data);
				getDout().flush();
				inputcount++;
			}else {
				break;
			}
		}
	}
	public KLALBBlock packBlock() throws IOException {
		byte[]dat=new byte[Consts.BLOCKSIZE];
		int len=getDin().read(dat);
		
		KLALBBlock pb=new KLALBBlock();
		pb.cuid=cuid;
		pb.number=outputcount++;
		if(len==-1) {
			pb.data=null;
		}else if(len==dat.length){
		pb.data=dat;
		}else {
			pb.data=Arrays.copyOf(dat, len);
		}
		return pb;
	}
	private BlockingDeque<KLALBBlock>sendDeque=new LinkedBlockingDeque<>();
	public BlockingDeque<KLALBBlock> getSendDeque() {
		return sendDeque;
	}
	private volatile int pcu=0;
	private Consumer<LocalTCPConnection> hook;
	public void setPeerCacheUsed(int cacheused) {
		pcu= cacheused;
	}
	public int getPeerCacheUsed() {
		return pcu;
	}
	public Consumer<LocalTCPConnection> getRSTHook() {
		return hook;
	}
	public void setRSTHook(Consumer<LocalTCPConnection> tc) {
		this.hook=tc;
	}
}