KLALB/src/org/kne/cloud/network/klalb/RemoteTCPConnection.java

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.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.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.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);
		}
	}
	
	@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());
		inm = new InputMetre(connect.getInputStream(),tunnel.getIM());
		}else {
			outm=connect.getOutputStream();
			inm=connect.getInputStream();
		}
		
		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);
	
	}
	@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;
	public void setDelayAvg(long delay) {
		if(odelay==Long.MAX_VALUE) {
			odelay=delay;
			setDelay(odelay);
		}else {
			odelay=(delay+odelay*100)/101;
			setDelay(odelay);
		}
	}
	
	@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<>();

	public Queue<KLALBBlock> getNDSendDeque() {
		return NDsendDeque;
	}
	private Queue<KLALBBlock> resendDeque = new ConcurrentLinkedQueue<>();

	public Queue<KLALBBlock> getReSendDeque() {
		return sendDeque;
	}
	private Queue<KLALBBlock> sendDeque = new ConcurrentLinkedQueue<>();

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

}