V1.0

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

@@ -1,31 +1,67 @@
 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.Random;
 import java.util.UUID;
 import java.util.concurrent.BlockingDeque;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.zip.GZIPInputStream;
+import java.util.zip.GZIPOutputStream;
 
 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 {
-		dout = new DataOutputStream(new OutputMetre( connect.getOutputStream()));
-		din = new DataInputStream(new InputMetre(connect.getInputStream()));
+		
+	
+	}
+	protected void initIOs() throws IOException {
+		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(new AddOutputStream(new BufferedOutputStream(outm)) );
+		din = new DataInputStream(new AddInputStream(new BufferedInputStream(inm)));
+	
 	}
-
 	public RemoteTCPConnection(Tunnel t, Socket s) throws UnknownHostException, IOException {
 		super(s);
-		this.tunnel = t;
+		tunnel=t;
+		initIOs();
 		s.setSoTimeout(Consts.SO_TIMEOUT);
 	}
 
@@ -52,11 +88,25 @@ public class RemoteTCPConnection extends TCPConnection {
 
 	}
 
+	@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 {
 		dout.writeLong(data.sn);
-		if(data.sn!=0) {
 		dout.writeLong(data.cuid);
-		}
 		dout.writeLong(data.number);
 		if (data.number > 0) {
 			if (data.data == null) {
@@ -83,17 +133,13 @@ public class RemoteTCPConnection extends TCPConnection {
 
 		dout.flush();
 
-		System.out.println("SEND:" + data);
+		//System.out.println("SEND:" + data);
 	}
 
 	public KLALBBlock receiveBlock() throws IOException {
 		KLALBBlock klb = new KLALBBlock();
 		klb.sn = din.readLong();
-		if(klb.sn!=0) {
 		klb.cuid = din.readLong();
-		}else {
-			klb.cuid=0;
-		}
 		klb.number = din.readLong();
 		if (klb.number > 0) {
 			int size = din.readInt();
@@ -119,7 +165,7 @@ public class RemoteTCPConnection extends TCPConnection {
 			klb.cacheused = din.readInt();
 		}
 
-		System.out.println("RECEIVE:" + klb);
+		//System.out.println("RECEIVE:" + klb);
 		return klb;
 	}
 
@@ -140,4 +186,5 @@ public class RemoteTCPConnection extends TCPConnection {
 	public BlockingDeque<KLALBBlock> getSendDeque() {
 		return sendDeque;
 	}
+
 }