package org.kne.cloud.network.klalb;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.StreamCorruptedException;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Comparator;
import java.util.Queue;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.zip.DataFormatException;
import java.util.zip.Deflater;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;
import java.util.zip.Inflater;

import org.kne.cloud.network.mport.IPPort;
import org.kne.io.AddInputStream;
import org.kne.io.AddOutputStream;

public class RemoteTCPConnection extends TCPConnection {

	public RemoteTCPConnection(Tunnel t) throws UnknownHostException, IOException {
		this(t, t.connectClientSocket());
		if (tunnel != null) {
		t.getRtcs().add(this);
		}
	}
	AtomicLong om=new AtomicLong(0);
	AtomicLong im=new AtomicLong(0);
	long om1=0;
	long im1=0;
	private volatile long ups,downs;
	private volatile double upsm,downsm;
	@Override
	protected void initIO() throws IOException {
		
	
	}
	
	protected void initIOs() throws IOException {
		connect.setTrafficClass(0x10);
		connect.setTcpNoDelay(true);
		OutputStream outm=null;
		InputStream inm=null;
		if(tunnel!=null) {
		outm = new OutputMetre( connect.getOutputStream(),tunnel.getOM(),om);
		inm = new InputMetre(connect.getInputStream(),tunnel.getIM(),im);
		}else {
			outm = new OutputMetre( connect.getOutputStream(),om);
			inm = new InputMetre(connect.getInputStream(),im);
		}
		
		new DataOutputStream(outm).writeShort(59649);//59649
		outm.flush();
			int val=new DataInputStream(inm).readShort()&0xffff;
			if(val!=59649) {
			throw new StreamCorruptedException();
		}
		
		
		dout = new DataOutputStream(outm);
		din = new DataInputStream(inm);
	
	}
	public long getUps() {
		return ups;
	}

	public long getDowns() {
		return downs;
	}

	public AtomicLong getOm() {
		return om;
	}

	public AtomicLong getIm() {
		return im;
	}
	private volatile long ptime=System.nanoTime();
	public void updateTraffics() {
		long ctime=System.nanoTime();
		long det=ctime-ptime;
		ptime=ctime;
		ups= (om.get()-om1)*1000000000/det;
		downs= (im.get()-im1)*1000000000/det;
		/*
		if(ndbg==null)
		initcdbg();
			ndbg.println((om.get()-om1)+ (im.get()-im1)+","+delay);
		*/
		om1=om.get();
		im1=im.get();
		if(ups>upsm) {
			upsm=ups;
		}else {
			upsm=(upsm*1000+ups)/1001;
		}
		if(downs>downsm) {
			downsm=downs;
		}else {
			downsm=(downsm*1000+downs)/1001;
		}
		//predictLatency();
		//System.out.println(tunnel+"//// "+upsm+" "+downsm+" "+getPredictedLatency());
	}
	private double latency;
	public double getPredictedLatency() {
		return latency;
	}
	public void predictLatency() {
		AtomicLong data=new AtomicLong(0);
		sendDeque.forEach((e)->{
			if(e.data!=null) {
				data.addAndGet(e.data.length);
			}
		});
		latency= delay/2000000.0+data.get()*1000.0/upsm;
	}

	@Override
	public String toString() {
		return "RemoteTCPConnection [tunnel=" + tunnel + "]";
	}

	public RemoteTCPConnection(Tunnel t, Socket s) throws UnknownHostException, IOException {
		super(s);
		tunnel=t;
		initIOs();
		s.setSoTimeout(Consts.SO_TIMEOUT);
	}

	private long delay = Long.MAX_VALUE;
	private Tunnel tunnel;

	public Tunnel getTunnel() {
		return tunnel;
	}

	public void setTunnel(Tunnel tunnel) {
		this.tunnel = tunnel;
	}

	public long getDelay() {
		return delay;
	}

	public void setDelay(long delay) {
		this.delay = delay;
		if (tunnel != null) {
			tunnel.setDelay(delay);
		}

	}
	long odelay=Long.MAX_VALUE;
	@Override
	public void close() {
		if (tunnel != null) {
		tunnel.getRtcs().remove(this);
		}
		super.close();
	}

	@Override
	protected void finalize() throws Throwable {
		if (tunnel != null) {
		tunnel.getRtcs().remove(this);
		}
		super.finalize();
	}
	public void sendBlock(KLALBBlock data) throws IOException {
		sendBlock(data,true);
	}
	public void sendBlock(KLALBBlock data,boolean isflush) throws IOException {
		
		dout.writeInt(data.cuid);
		dout.writeLong(data.number);
		if (data.number > 0) {
			if (data.data == null) {
				dout.writeShort(-1);
			} else {
				/*Deflater def = new Deflater(Deflater.BEST_COMPRESSION, true);
		        def.setInput(data.data);
		        def.finish();
		        byte[] b = new byte[(int)(data.data.length * 1.1D) + 64];
		        int nsize = def.deflate(b);
		        def.end();
		        this.dout.writeShort(nsize);
		        this.dout.write(b, 0, nsize);*/
				dout.writeShort(data.data.length);
				dout.write(data.data);
				//System.out.println(Arrays.toString(data.data));
			}
		} else if (data.number == 0) {
			dout.write(data.command);
			if(data.command!=0&&data.command!=1) {
			dout.writeLong(data.sn);
			}else {
				
			}
			switch (data.command) {
			case 0:
			case 1:
				dout.writeLong(data.pingtime);
				break;
			case 2:
				dout.writeUTF(data.lservice);
				dout.writeUTF(data.lipport.toString());
				break;
			default:
				break;
			}
		} else {
			dout.writeInt(data.cacheused);
		}

		if(isflush)
		dout.flush();

		System.out.println("SEND:" + data);
	}

	public KLALBBlock receiveBlock() throws IOException, DataFormatException {
		KLALBBlock klb = new KLALBBlock();
		klb.cuid = din.readInt();
		klb.number = din.readLong();
		if (klb.number > 0) {
			int size = din.readShort();
			if (size == -1) {
				klb.data = null;
			} else {
				byte[] d = new byte[size];
				din.readFully(d);
				 /*Inflater in = new Inflater(true);
			        in.setInput(d);
			        byte[] b = new byte[8192];
			        int nsize = in.inflate(b);
			        in.end();
			        if (nsize == b.length) {
			          klb.data = b;
			        } else {
			          klb.data = Arrays.copyOf(b, nsize);
			        } */
				//System.out.println(Arrays.toString(d));
				klb.data = d;
			}
		} else if (klb.number == 0) {
			klb.command = din.read();
			if(klb.command!=0&&klb.command!=1) {
		klb.sn = din.readLong();
			}else {
				klb.sn=0;
			}
			switch (klb.command) {
			case 0:
			case 1:
				klb.pingtime = din.readLong();
				break;
			case 2:
				klb.lservice = din.readUTF();
				klb.lipport = new IPPort(din.readUTF());
			break;
				default:
					break;
			}
		} else {
			klb.cacheused = din.readInt();
		}

		System.out.println("RECEIVE:" + klb);
		return klb;
	}

	private volatile long time = System.nanoTime();

	public boolean checkPingTime() {
		long cu = System.nanoTime();
		if (cu - time > Consts.PINGTIMENS) {
			time = cu;
			return true;
		} else {
			return false;
		}
	}
	
	private Queue<KLALBBlock> NDsendDeque = new ConcurrentLinkedQueue<KLALBBlock>();

	public Queue<KLALBBlock> getNDSendDeque() {
		return NDsendDeque;
	}
	//private Queue<KLALBBlock> sendDeque = new ConcurrentLinkedQueue<KLALBBlock>();
	private Queue<KLALBBlock> sendDeque = new PriorityBlockingQueue<KLALBBlock>(2, new Comparator<KLALBBlock>() {

		@Override
		public int compare(KLALBBlock o1, KLALBBlock o2) {
			if(o1.resend>o2.resend) {
				return -1;
				}else if(o1.resend<o2.resend) {
					return 1;
				}else {
			return 0;
			}
		}
	});

	public Queue<KLALBBlock> getSendDeque() {
		return sendDeque;
	}

	public void handshake(int x) throws IOException {
		for(int i=0;i<x;i++) {
			getDout().write(0);
			getDout().flush();
			int v=getDin().read();
			if(v==-1) {
				throw new EOFException();
			}
			if(v!=0) {
				throw new StreamCorruptedException();
			}
		}
		
	}

}