KLALB UDP TEST

.classpath

@@ -2,8 +2,13 @@
 <classpath>
 	<classpathentry kind="src" path="src"/>
 	<classpathentry combineaccessrules="false" kind="src" path="/KNElib"/>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/jdk-19">
+		<attributes>
+			<attribute name="module" value="true"/>
+		</attributes>
+	</classpathentry>
 	<classpathentry kind="lib" path="lib/gson-2.1.jar"/>
 	<classpathentry kind="lib" path="lib/javassist.jar" sourcepath="C:/Users/ADMINI~1/AppData/Local/Temp/1/.org.sf.feeling.decompiler1686963856070/source/javassist-3.29.2-GA-sources.jar"/>
-	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/jdk-19"/>
+	<classpathentry kind="lib" path="lib/ini4j.jar"/>
 	<classpathentry kind="output" path="bin"/>
 </classpath>

kserver.ini

@@ -1,3 +1,3 @@
-virtualip=67c:72ce:765e:4db1:a02e:92fa:1959:29eb
+virtualip=cefe:49b8:f837:4b00:b9cd:dbdd:c299:349
 bind=0.0.0.0:4569
-local=127.0.0.1:36555
+local=127.0.0.1:5212

linetable - 副本.txt

@@ -15,5 +15,4 @@ frp.freefrp.net:49965
 frp1.freefrp.net:49965
 frp2.freefrp.net:49965
 frp4.freefrp.net:49965
-192.168.0.233:4569
-192.168.1.233:4569
+cn-he-plc-2.openfrp.top:4569

linetable.txt

@@ -1,20 +1,2 @@
-43.248.189.107:65529
-cn-bj-bgp-3.openfrp.top:65529
-180.76.147.250:65529
-cn-ah-dx-1.natfrp.cloud:65529
-cn-nn-dx-1.natfrp.cloud:65529
-cn-wh-dx-1.natfrp.cloud:65529
-cn-zz-bgp-10.natfrp.cloud:23330
-cn-zz-bgp-7.natfrp.cloud:33336
-43.143.109.64:49965
-frp.104300.xyz:49965
-us.afrps.cn:49966
-hk.afrps.cn:49966
-la.afrps.cn:49966
-frp.freefrp.net:49965
-frp1.freefrp.net:49965
-frp2.freefrp.net:49965
-frp4.freefrp.net:49965
-192.168.0.233:4569
-192.168.1.233:4569
-cn-he-plc-2.openfrp.top:4569
+{UDP}192.168.1.233:4569
+{UDP}192.168.0.233:4569

src/org/kne/cloud/network/DefaultServerSocketFactory.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.IOException;
 import java.net.InetAddress;
@@ -8,6 +8,12 @@ import javax.net.ServerSocketFactory;
 
 public class DefaultServerSocketFactory extends ServerSocketFactory {
 	
+
+	@Override
+	public ServerSocket createServerSocket() throws IOException {
+		return new ServerSocket();
+	}
+
 	@Override
 	public ServerSocket createServerSocket(int port) throws IOException {
 		return new ServerSocket(port);

src/org/kne/cloud/network/DefaultSocketFactory.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.IOException;
 import java.net.InetAddress;
@@ -7,7 +7,9 @@ import java.net.UnknownHostException;
 import javax.net.SocketFactory;
 
 public class DefaultSocketFactory extends SocketFactory {
-	public Socket createSocket() {
+
+	@Override
+	public Socket createSocket() throws IOException {
 		return new Socket();
 	}
 

src/org/kne/cloud/network/FilterServerSocket.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.IOException;
 import java.net.InetAddress;

src/org/kne/cloud/network/FilterSocket.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.FilterInputStream;
 import java.io.IOException;

src/org/kne/cloud/network/HTTPDetectorItem.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.DataInputStream;
 import java.io.IOException;

src/org/kne/cloud/network/HTTPSDetectorItem.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.DataInputStream;
 import java.io.IOException;

src/org/kne/cloud/network/HostPortMap.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.util.HashMap;
 import java.util.LinkedHashMap;

src/org/kne/cloud/network/KLALBDetectorItem.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.DataInputStream;
 import java.io.IOException;

src/org/kne/cloud/network/MultipurposeSocketAddress.java

@@ -1,7 +1,8 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.IOException;
 import java.io.Serializable;
+import java.net.DatagramSocket;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
@@ -19,20 +20,19 @@ public class MultipurposeSocketAddress implements Serializable{
 	/**
 	 * 
 	 */
-	private static Map<String,SocketFactory> socketFactoryRegister=new HashMap<>();
-	private static Map<String,ServerSocketFactory> serverSocketFactoryRegister=new HashMap<>();
+	private static Map<String,SocketType> socketTypeRegister=new HashMap<>();
+	
+	
 	static {
-		socketFactoryRegister.put("TCP", new DefaultSocketFactory());
-		serverSocketFactoryRegister.put("TCP", new DefaultServerSocketFactory());
+		socketTypeRegister.put("TCP", new SocketType(new DefaultSocketFactory(), new DefaultServerSocketFactory()));
+		socketTypeRegister.put("UDP", new SocketType(new DefaultDatagramSocketFactory(),new DefaultDatagramServerSocketFactory()));
 	}
 	
-	public static Map<String, SocketFactory> getSocketFactoryRegister() {
-		return socketFactoryRegister;
-	}
-	public static Map<String, ServerSocketFactory> getServerSocketFactoryRegister() {
-		return serverSocketFactoryRegister;
-	}
 
+	
+	public static Map<String, SocketType> getSocketTypeRegister() {
+		return socketTypeRegister;
+	}
 	private static final long serialVersionUID = 1L;
 	private String type;
 	private String host;
@@ -108,6 +108,9 @@ public class MultipurposeSocketAddress implements Serializable{
 	public int getPort() {
 		return port;
 	}
+	public String getType() {
+		return type;
+	}
 	@Override
 	public String toString() {
 		StringBuilder sb=new StringBuilder();
@@ -125,35 +128,99 @@ public class MultipurposeSocketAddress implements Serializable{
 		return sb.toString();
 	}
 	public Socket connectSocket(InetAddress bindip,int bindport,int timeout) throws UnknownHostException, IOException {
-		Socket s=socketFactoryRegister.get(type).createSocket();
+		SocketFactory sf=socketTypeRegister.get(type).getSocketFactory();
+		if(sf==null) {
+			throw new UnsupportedOperationException("Socket Unsupported");
+		}
+		Socket s=sf.createSocket();
 		s.bind(new InetSocketAddress(bindip, bindport));
 		s.connect(new InetSocketAddress(host, port),timeout);
 		return s;
 	}
 	public Socket connectSocket(InetAddress bindip,int bindport) throws UnknownHostException, IOException {
-		Socket s=socketFactoryRegister.get(type).createSocket();
+		SocketFactory sf=socketTypeRegister.get(type).getSocketFactory();
+		if(sf==null) {
+			throw new UnsupportedOperationException("Socket Unsupported");
+		}
+		Socket s=sf.createSocket();
 		s.bind(new InetSocketAddress(bindip, bindport));
 		s.connect(new InetSocketAddress(host, port));
 		return s;
 	}
 	public Socket connectSocket() throws UnknownHostException, IOException {
-		Socket s=socketFactoryRegister.get(type).createSocket();
+		SocketFactory sf=socketTypeRegister.get(type).getSocketFactory();
+		if(sf==null) {
+			throw new UnsupportedOperationException("Socket Unsupported");
+		}
+		Socket s=sf.createSocket();
 		s.connect(new InetSocketAddress(host, port));
 		return s;
 	}
 	public Socket connectSocket(int timeout) throws UnknownHostException, IOException {
-		Socket s=socketFactoryRegister.get(type).createSocket();
+		SocketFactory sf=socketTypeRegister.get(type).getSocketFactory();
+		if(sf==null) {
+			throw new UnsupportedOperationException("Socket Unsupported");
+		}
+		Socket s=sf.createSocket();
 		s.connect(new InetSocketAddress(host, port),timeout);
 		return s;
 	}
 	
 	public ServerSocket listenServerSocket() throws UnknownHostException, IOException {
-		ServerSocket sk=serverSocketFactoryRegister.get(type).createServerSocket(port, 50, InetAddress.getByName(host));
+		ServerSocketFactory srf=socketTypeRegister.get(type).getServerSocketFactory();
+		if(srf==null) {
+			throw new UnsupportedOperationException("ServerSocket Unsupported");
+		}
+		ServerSocket sk=srf.createServerSocket(port, 50, InetAddress.getByName(host));
 		return sk;
 	}
 	public ServerSocket listenServerSocket(int backlog) throws UnknownHostException, IOException {
-		ServerSocket sk=serverSocketFactoryRegister.get(type).createServerSocket(port, backlog, InetAddress.getByName(host));
+		ServerSocketFactory srf=socketTypeRegister.get(type).getServerSocketFactory();
+		if(srf==null) {
+			throw new UnsupportedOperationException("ServerSocket Unsupported");
+		}
+		ServerSocket sk=srf.createServerSocket(port, backlog, InetAddress.getByName(host));
+		return sk;
+	}
+	public boolean isStream() {
+		return checkIsStream(type);
+	}
+	public static boolean checkIsStream(String type2) {
+		return socketTypeRegister.get(type2).isStream();
+	}
+	public DatagramSocket connectDatagramSocket(InetAddress bindip,int bindport) throws IOException {
+		DatagramSocketFactory dgs=socketTypeRegister.get(type).getDatagramSocketFactory();
+		if(dgs==null) {
+			throw new UnsupportedOperationException("DatagramSocket Unsupported");
+		}
+		DatagramSocket dgd=dgs.createSocket();
+		dgd.bind(new InetSocketAddress(bindip, bindport));
+		dgd.connect(new InetSocketAddress(host, port));
+		return dgd;
+	}
+	public DatagramSocket connectDatagramSocket() throws IOException {
+		DatagramSocketFactory dgs=socketTypeRegister.get(type).getDatagramSocketFactory();
+		if(dgs==null) {
+			throw new UnsupportedOperationException("DatagramSocket Unsupported");
+		}
+		DatagramSocket dgd=dgs.createSocket();
+		dgd.connect(new InetSocketAddress(host, port));
+		return dgd;
+	}
+	public DatagramServerSocket listenDatagramServerSocket() throws UnknownHostException, IOException {
+		DatagramServerSocketFactory srf=socketTypeRegister.get(type).getDatagramServerSocketFactory();
+		if(srf==null) {
+			throw new UnsupportedOperationException("DatagramServerSocket Unsupported");
+		}
+		DatagramServerSocket sk=srf.createDatagramServerSocket(port, 50, InetAddress.getByName(host));
+		return sk;
+	}
+	public DatagramServerSocket listenDatagramServerSocket(int backlog) throws UnknownHostException, IOException {
+		DatagramServerSocketFactory srf=socketTypeRegister.get(type).getDatagramServerSocketFactory();
+		if(srf==null) {
+			throw new UnsupportedOperationException("DatagramServerSocket Unsupported");
+		}
+		DatagramServerSocket sk=srf.createDatagramServerSocket(port, backlog, InetAddress.getByName(host));
 		return sk;
 	}
-	
 }

src/org/kne/cloud/network/NetworkService.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 public interface NetworkService {
 	public void listen(MultipurposeSocketAddress msa);

src/org/kne/cloud/network/PortMultiUse.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.File;
 import java.io.IOException;

src/org/kne/cloud/network/PortRelay.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
@@ -12,12 +12,12 @@ import java.util.*;
 public class PortRelay {
 	private int port;
 	private HostPortMap services;
-	private TCPListener ssc;
+	private SocketListener ssc;
 	public PortRelay(int port, HostPortMap services) throws IOException {
 		this.services = services;
 		this.port = port;
-		MultipurposeSocketAddress.getServerSocketFactoryRegister().put("ProtocolDetectorServerSocket", new ProtocolDetectorServerSocketFactory());
-	ssc=new TCPListener(new MultipurposeSocketAddress("ProtocolDetectorServerSocket", "0.0.0.0", port));
+		MultipurposeSocketAddress.getSocketTypeRegister().put("ProtocolDetectorServerSocket",new SocketType(null, new ProtocolDetectorServerSocketFactory()) );
+	ssc=new SocketListener(new MultipurposeSocketAddress("ProtocolDetectorServerSocket", "0.0.0.0", port));
 	}
 	public void start() throws IOException {
 		ssc.setCon((s)->{
@@ -55,7 +55,6 @@ public class PortRelay {
 			}
 			
 		});
-		ssc.open();
 		System.out.println("已打开端口：" + port);
 	}
 	

src/org/kne/cloud/network/Protocol.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.util.Objects;
 

src/org/kne/cloud/network/ProtocolDetector.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.IOException;
 import java.io.InputStream;

src/org/kne/cloud/network/ProtocolDetectorItem.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.InputStream;
 import java.util.function.Function;

src/org/kne/cloud/network/ProtocolDetectorServerSocket.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.IOException;
 import java.io.InputStream;

src/org/kne/cloud/network/ProtocolDetectorServerSocketFactory.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.IOException;
 import java.net.InetAddress;
@@ -17,9 +17,13 @@ public class ProtocolDetectorServerSocketFactory extends DefaultServerSocketFact
 		this.pd = pd;
 	}
 
+	@Override
+	public ServerSocket createServerSocket() throws IOException {
+		return new ProtocolDetectorServerSocket(super.createServerSocket());
+	}
+
 	@Override
 	public ServerSocket createServerSocket(int port) throws IOException {
-		// TODO 自动生成的方法存根
 		return new ProtocolDetectorServerSocket( super.createServerSocket(port),pd);
 	}
 
@@ -33,13 +37,11 @@ public class ProtocolDetectorServerSocketFactory extends DefaultServerSocketFact
 
 	@Override
 	public ServerSocket createServerSocket(int port, int backlog) throws IOException {
-		// TODO 自动生成的方法存根
 		return new ProtocolDetectorServerSocket(super.createServerSocket(port, backlog),pd);
 	}
 
 	@Override
 	public ServerSocket createServerSocket(int port, int backlog, InetAddress ifAddress) throws IOException {
-		// TODO 自动生成的方法存根
 		return new ProtocolDetectorServerSocket(super.createServerSocket(port, backlog, ifAddress),pd);
 	}
 

src/org/kne/cloud/network/ProtocolDetectorSocket.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.BufferedInputStream;
 import java.io.IOException;

src/org/kne/cloud/network/ProtocolStack.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.util.Stack;
 

src/org/kne/cloud/network/Proxy.java

@@ -1,5 +1,7 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
+import java.io.Closeable;
+import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.util.HashMap;
 import java.util.Map;
@@ -8,9 +10,13 @@ import java.util.function.Consumer;
 import javax.net.ServerSocketFactory;
 import javax.net.SocketFactory;
 
-public class ProxyProfileEntry {
+public abstract class Proxy implements Closeable{
 	private static Map<String,NetworkService> register=new HashMap<>();
 	public static Map<String, NetworkService> getRegister() {
 		return register;
 	}
+	
+	
+	
+	
 }

src/org/kne/cloud/network/RDPDetectorItem.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.DataInputStream;
 import java.io.IOException;

src/org/kne/cloud/network/SSHDetectorItem.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.DataInputStream;
 import java.io.IOException;

src/org/kne/cloud/network/ServiceElement.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.net.UnknownHostException;
 import java.util.HashMap;

src/org/kne/cloud/network/ServiceToSocketProxy.java

@@ -1,18 +1,24 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
+import java.io.IOException;
 import java.util.function.Consumer;
 
-public class ServiceToSocketProxyProfileEntry extends ProxyProfileEntry{
+public class ServiceToSocketProxy extends Proxy{
 	public NetworkService getSrc() {
 		return src;
 	}
 	public MultipurposeSocketAddress getDes() {
 		return des;
 	}
-	public ServiceToSocketProxyProfileEntry(String l, String r) {
+	public ServiceToSocketProxy(String l, String r) {
 		src=getRegister().get(l.substring(1, l.length()-1));
 		des=new MultipurposeSocketAddress(r);
 	}
 	private NetworkService src;
 	private MultipurposeSocketAddress des;
+	@Override
+	public void close() throws IOException {
+		// TODO 自动生成的方法存根
+		
+	}
 }

src/org/kne/cloud/network/SocketBridge.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -6,33 +6,37 @@ import java.net.*;
 
 import org.kne.io.Task;
 public class SocketBridge extends Task{
-	private Socket a;
-	private Socket b;
-	public StreamBridge getSab() {
-		return sab;
+	protected Socket a;
+	protected Socket b;
+	protected StreamBridge bridgeAB;
+	protected StreamBridge bridgeBA;
+	public StreamBridge getBridgeAB() {
+		return bridgeAB;
 	}
-	public StreamBridge getSba() {
-		return sba;
+	public StreamBridge getBridgeBA() {
+		return bridgeBA;
 	}
-	private StreamBridge sab;
-	private StreamBridge sba;
 	public SocketBridge(Socket a, Socket b) throws IOException {
 		super();
 		this.a = a;
 		this.b = b;
-		sab = new StreamBridge(getAIN(), getBOUT());
-		sba = new StreamBridge(getBIN(), getAOUT());
+		createStreamBridge();
+	}
+	protected void createStreamBridge() throws IOException {
+		bridgeAB = new StreamBridge(a.getInputStream(), b.getOutputStream());
+		bridgeBA = new StreamBridge(b.getInputStream(), a.getOutputStream());
 	}
 	@Override
 	protected void runTask() {
 		try {
-			sab.runAtNewThread("SocketBridge A->B thread");
-			sba.runAtNewThread("SocketBridge B->A thread");
-			sab.waitfortask();
-			sba.waitfortask();
+			bridgeAB.runAtNewThread("SocketBridge A->B thread");
+			bridgeBA.runAtNewThread("SocketBridge B->A thread");
+			bridgeAB.waitfortask();
+			bridgeBA.waitfortask();
 		} catch (Exception e) {
 			e.printStackTrace();
 		}finally {
+			//new Exception().printStackTrace();
 			try {
 				a.close();
 			} catch (IOException e) {
@@ -53,7 +57,7 @@ public class SocketBridge extends Task{
 	public Socket getB() {
 		return b;
 	}
-	protected OutputStream getBOUT() throws IOException {
+	/*protected OutputStream getBOUT() throws IOException {
 		return b.getOutputStream();
 	}
 	protected InputStream getBIN() throws IOException {
@@ -64,6 +68,6 @@ public class SocketBridge extends Task{
 	}
 	protected InputStream getAIN() throws IOException {
 		return a.getInputStream();
-	}
+	}*/
 	
 }

src/org/kne/cloud/network/SocketListener.java

@@ -1,22 +1,25 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
+import java.io.Closeable;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.ServerSocket;
 import java.net.Socket;
+import java.net.SocketException;
+import java.net.UnknownHostException;
 import java.util.function.Consumer;
 
 import javax.net.ServerSocketFactory;
 
-public class TCPListener {
-	private ServerSocket serverSocket;
+public class SocketListener implements Closeable,AutoCloseable{
+	protected ServerSocket serverSocket;
 	public ServerSocket getServerSocket() {
 		return serverSocket;
 	}
 
-	private volatile boolean flag=false;
+	private volatile boolean flag=true;
 	
-	private Consumer<Socket>con;
+	private volatile Consumer<Socket>con;
 	private Runnable r=new Runnable() {
 		@Override
 		public void run() {
@@ -24,8 +27,26 @@ public class TCPListener {
 				try {
 					Socket soce=serverSocket.accept();
 					ThreadTool.makeVThreadIfSupport("端口监听线程",()->{
+							if(con!=null) {
+								try {
 						con.accept(soce);
+								}catch(Exception e) {
+									e.printStackTrace();
+									try {
+										soce.close();
+									} catch (IOException er) {
+										er.printStackTrace();
+									}
+								}
+							}else {
+								try {
+									soce.close();
+								} catch (IOException e) {
+									e.printStackTrace();
+								}
+						}
 					}).start();
+				}catch(SocketException e) {
 				} catch (IOException e) {
 					e.printStackTrace();
 				}
@@ -34,14 +55,17 @@ public class TCPListener {
 	};
 	private MultipurposeSocketAddress multipurposeSocketAddress;
 	
-	public TCPListener(MultipurposeSocketAddress multipurposeSocketAddress) {
+	public SocketListener(MultipurposeSocketAddress multipurposeSocketAddress) throws  IOException {
 		this.multipurposeSocketAddress=multipurposeSocketAddress;
+			open();
 	}
-
-	public void open() throws IOException {
-		flag=true;
+	public SocketListener(ServerSocket tserverSocket) throws  IOException {
+		this.serverSocket=tserverSocket;
+			open();
+	}
+	protected void open() throws UnknownHostException, IOException  {
+		if(serverSocket==null)
 			serverSocket=multipurposeSocketAddress.listenServerSocket();
-		//servers=new ServerSocket(port);
 		new Thread(r).start();
 	}
 	public void close() {

src/org/kne/cloud/network/SocketToServiceProxy.java

@@ -1,10 +1,11 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
+import java.io.IOException;
 import java.net.Socket;
 import java.util.function.Consumer;
 
-public class SocketToServiceProxyProfileEntry extends ProxyProfileEntry {
-	public SocketToServiceProxyProfileEntry(String l, String r) {
+public class SocketToServiceProxy extends Proxy {
+	public SocketToServiceProxy(String l, String r) {
 		src=new MultipurposeSocketAddress(l);
 		des=getRegister().get(r.substring(1, r.length()-1)) ;
 	}
@@ -16,4 +17,9 @@ public class SocketToServiceProxyProfileEntry extends ProxyProfileEntry {
 	}
 	private MultipurposeSocketAddress src;
 	private NetworkService des;
+	@Override
+	public void close() throws IOException {
+		// TODO 自动生成的方法存根
+		
+	}
 }

src/org/kne/cloud/network/SocketToSocketProxy.java

@@ -0,0 +1,130 @@
+package org.kne.cloud.network;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.UnknownHostException;
+import java.util.Map;
+
+public class SocketToSocketProxy extends Proxy {
+	private SocketListener sl;
+	private DatagramSocketListener dsl;
+	private MultipurposeSocketAddress listen, cbind, defaultConnect;
+	private Map<String, MultipurposeSocketAddress> detectedConnect;
+
+	public SocketToSocketProxy(MultipurposeSocketAddress listen, MultipurposeSocketAddress connect) throws IOException {
+		this(listen, new MultipurposeSocketAddress(new InetSocketAddress(0)), connect);
+	}
+
+	public SocketToSocketProxy(String l, String r) throws IOException {
+		this(new MultipurposeSocketAddress(l), new MultipurposeSocketAddress(r));
+	}
+
+	public SocketToSocketProxy(MultipurposeSocketAddress listen, MultipurposeSocketAddress cbind,
+			MultipurposeSocketAddress defaultConnect, Map<String, MultipurposeSocketAddress> detectedConnect)
+			throws IOException {
+		this.listen = listen;
+		this.cbind = cbind;
+		this.defaultConnect = defaultConnect;
+		this.detectedConnect = detectedConnect;
+		open();
+	}
+
+	public SocketToSocketProxy(MultipurposeSocketAddress listen, MultipurposeSocketAddress cbind,
+			MultipurposeSocketAddress defaultConnect) throws IOException {
+		this(listen, cbind, defaultConnect,null);
+	}
+
+	private void open() throws IOException {
+		if (listen.isStream()) {
+			if (detectedConnect == null || detectedConnect.isEmpty()) {
+				sl = new SocketListener(listen);
+			} else {
+				sl = new SocketListener(new ProtocolDetectorServerSocket(listen.listenServerSocket()));
+			}
+			sl.setCon((sox) -> {
+				Socket sk = null;
+				try {
+					if (sox instanceof ProtocolDetectorSocket) {
+						ProtocolStack ps = ((ProtocolDetectorSocket) sox).getProtocolStack();
+						if (!ps.isEmpty()) {
+						MultipurposeSocketAddress pmsa = detectedConnect.get(ps.pop().getName());
+						if(pmsa!=null) {
+							sk = pmsa.connectSocket(InetAddress.getByName(cbind.getHost()), cbind.getPort());
+						}else {
+							sk = defaultConnect.connectSocket(InetAddress.getByName(cbind.getHost()), cbind.getPort());
+							
+						}
+						} else {
+							sk = defaultConnect.connectSocket(InetAddress.getByName(cbind.getHost()), cbind.getPort());
+						}
+					} else {
+						sk = defaultConnect.connectSocket(InetAddress.getByName(cbind.getHost()), cbind.getPort());
+					}
+					runBridge(sox,sk);
+				} catch (IOException e) {
+					e.printStackTrace();
+				} finally {
+					if (sk != null)
+						try {
+							sk.close();
+						} catch (IOException e) {
+							e.printStackTrace();
+						}
+					try {
+						sox.close();
+					} catch (IOException e) {
+						e.printStackTrace();
+					}
+				}
+			});
+		} else {
+			dsl = new DatagramSocketListener(listen);
+			dsl.setCon((dox) -> {
+				
+			});
+		}
+	}
+
+	protected void runBridge(Socket sk, Socket sox) throws IOException {
+		SocketBridge sb = new SocketBridge(sk, sox);
+		sb.run();
+	}
+
+	public MultipurposeSocketAddress getListen() {
+		return listen;
+	}
+
+	public MultipurposeSocketAddress getCbind() {
+		return cbind;
+	}
+
+	public MultipurposeSocketAddress getDefaultConnect() {
+		return defaultConnect;
+	}
+
+	@Override
+	public void close() throws IOException {
+		if (dsl != null)
+			dsl.close();
+		if (sl != null)
+			sl.close();
+	}
+
+	public InetAddress getListenAddress() {
+		if (sl != null) {
+			return sl.getServerSocket().getInetAddress();
+		} else {
+			return dsl.getDatagramServerSocket().getLocalAddress();
+		}
+	}
+
+	public int getListenPort() {
+		if (sl != null) {
+			return sl.getServerSocket().getLocalPort();
+		} else {
+			return dsl.getDatagramServerSocket().getLocalPort();
+		}
+	}
+}

src/org/kne/cloud/network/StreamBridge.java

@@ -1,15 +1,16 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.*;
 import java.util.Objects;
+import java.util.UUID;
 
 import org.kne.io.Task;
 
 public class StreamBridge extends Task{
-	private InputStream in;
-	private OutputStream out;
-	private int blocksize=65535;
-	private long delay=0;
+	protected InputStream in;
+	protected OutputStream out;
+	protected int blocksize=65535;
+	protected long delay=0;
 	public StreamBridge(InputStream in, OutputStream out) {
 		super();
 		Objects.requireNonNull(in);
@@ -20,12 +21,16 @@ public class StreamBridge extends Task{
 	
 	@Override
 	public void runTask() {
+		//FileOutputStream fos=null;
 		try {
+			//fos=new FileOutputStream(UUID.randomUUID()+".txt");
 		int v=-1;
 		byte[]b=new byte[blocksize];
 			while ((v=in.read(b))!=-1) {
+			//	fos.write(b, 0, v);
 				out.write(b,0,v);
 				out.flush();
+				if(delay>0)
 				try {
 					Thread.sleep(delay);
 				} catch (InterruptedException e) {
@@ -36,13 +41,21 @@ public class StreamBridge extends Task{
 			// TODO 自动生成的 catch 块
 			e.printStackTrace();
 		}finally {
+			/*if(fos!=null)
 				try {
+					fos.close();
+				} catch (IOException e1) {
+					e1.printStackTrace();
+				}*/
+			try {
+				if(in!=null)
 				in.close();
 			} catch (IOException e) {
 				// TODO 自动生成的 catch 块
 				e.printStackTrace();
 			}
 			try {
+				if(in!=null)
 				out.close();
 			} catch (IOException e) {
 				// TODO 自动生成的 catch 块

src/org/kne/cloud/network/ThreadTool.java

@@ -1,10 +1,10 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.lang.reflect.Method;
 
 public class ThreadTool {
 	public static boolean first=true;
-	public static boolean forceD=true;
+	public static boolean forceD=false;
 	public static Thread makeVThreadIfSupport(String name,Runnable r) {
 		if(forceD)
 			return new Thread(r, name);
@@ -21,7 +21,7 @@ public class ThreadTool {
 		}catch(Throwable e) {
 			//e.printStackTrace();
 			if(first) {
-			System.out.println("请使用java19以上版本并开启--enable-preview选项以提高性能！");
+			System.out.println("提示：请使用java21以上版本以提高本软件的数据转发性能！");
 			first=false;
 			}
 			return new Thread(r, name);
@@ -46,7 +46,7 @@ public class ThreadTool {
 		}catch(Throwable e) {
 			//e.printStackTrace();
 			if(first) {
-			System.out.println("请使用java19以上版本以提高性能！");
+			System.out.println("提示：请使用java21以上版本以提高本软件的数据转发性能！");
 			first=false;
 			}
 			Thread rt=new Thread(r, name);
@@ -54,4 +54,14 @@ public class ThreadTool {
 			return rt;
 		}
 	}
+
+	public static Thread makePThreadIfSupport(String name, Runnable r) {
+		Thread rt=new Thread(r, name);
+		return rt;
+	}
+	public static Thread makePDaemonThreadIfSupport(String name, Runnable r) {
+		Thread rt=new Thread(r, name);
+		rt.setDaemon(true);
+		return rt;
+	}
 }

src/org/kne/cloud/network/VirtualServerSocket.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.IOException;
 import java.lang.reflect.Constructor;
@@ -13,8 +13,11 @@ import javax.sql.rowset.RowSetMetaDataImpl;
 
 public abstract class VirtualServerSocket extends ServerSocket {
 
-	public VirtualServerSocket(SocketImpl si) throws IOException {
+	private VirtualSocketImpl virtualImpl;
+
+	public VirtualServerSocket(VirtualSocketImpl si) throws IOException {
 		super(si);
+		this.virtualImpl=si;
 		/*try {
 		Class<ServerSocket> c=ServerSocket.class;
 		Field con	=c.getDeclaredField("impl");
@@ -45,6 +48,10 @@ public abstract class VirtualServerSocket extends ServerSocket {
 		}*/
 	}
 
+	public VirtualSocketImpl getVirtualImpl() {
+		return virtualImpl;
+	}
+
 	@Override
 	public abstract VirtualSocket accept() throws IOException ;
 

src/org/kne/cloud/network/VirtualSocket.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.FilterInputStream;
 import java.io.IOException;
@@ -13,21 +13,25 @@ import java.nio.channels.SocketChannel;
 
 public abstract class VirtualSocket extends Socket {
 	
-	private VirtualSocketImpl si;
+	private VirtualSocketImpl virtualImpl;
+
+	public VirtualSocketImpl getVirtualImpl() {
+		return virtualImpl;
+	}
 
 	public VirtualSocket(VirtualSocketImpl si) throws SocketException {
 		super(si);
-		this.si=si;
+		this.virtualImpl=si;
 	}
 
 	@Override
 	public InputStream getInputStream() throws IOException {
-		return si.getInputStream();
+		return virtualImpl.getInputStream();
 	}
 
 	@Override
 	public OutputStream getOutputStream() throws IOException {
-		return si.getOutputStream();
+		return virtualImpl.getOutputStream();
 	}
 
 	

src/org/kne/cloud/network/VirtualSocketImpl.java

@@ -1,4 +1,4 @@
-package org.kne.cloud.network.mport;
+package org.kne.cloud.network;
 
 import java.io.IOException;
 import java.io.InputStream;

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

@@ -1,19 +1,25 @@
 package org.kne.cloud.network.klalb;
 
 import java.io.DataInput;
+import java.io.DataInputStream;
 import java.io.DataOutput;
+import java.io.DataOutputStream;
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
 
-public class ACKTPacket extends KLALBPacket {
+public class ACKTPacket extends KLALBPacket implements PortPacket {
 	private int sport,dport;
 	private long number;
 	private boolean avaliable;
-	public ACKTPacket(int sport,int dport,long number,boolean avaliable) {
+	private int sendcount;
+
+	public ACKTPacket(int sport,int dport,long number,boolean avaliable,int sendcount) {
 		super(ACKT);
 		this.sport=sport;
 		this.dport=dport;
 		this.number=number;
 		this.avaliable=avaliable;
+		this.sendcount=sendcount;
 	}
 
 	public ACKTPacket() {
@@ -31,7 +37,7 @@ public class ACKTPacket extends KLALBPacket {
 
 	@Override
 	public long getLength() {
-		return super.getLength()+17;
+		return super.getLength()+18;
 	}
 
 	public int getDport() {
@@ -46,21 +52,67 @@ public class ACKTPacket extends KLALBPacket {
 		return avaliable;
 	}
 	
+
+	public int getSendcount() {
+		return sendcount;
+	}
+
+	 private final byte[] writeBuffer = new byte[18];
 	@Override
 	protected void writeToStream(DataOutput dto) throws IOException {
 		super.writeToStream(dto);
-		dto.writeInt(sport);
+		/*dto.writeInt(sport);
 		dto.writeInt(dport);
 		dto.writeLong(number);
-		dto.writeBoolean(avaliable);
-	}
+		dto.writeByte(sendcount);
+		dto.writeBoolean(avaliable);*/
+		writeBuffer[0] = (byte)(sport >>> 24);
+        writeBuffer[1] = (byte)(sport >>> 16);
+        writeBuffer[2] = (byte)(sport >>>  8);
+        writeBuffer[3] = (byte)(sport >>>  0);
         
+    	writeBuffer[4] = (byte)(dport >>> 24);
+        writeBuffer[5] = (byte)(dport >>> 16);
+        writeBuffer[6] = (byte)(dport >>>  8);
+        writeBuffer[7] = (byte)(dport >>>  0);
+
+        writeBuffer[8] = (byte)(number >>> 56);
+        writeBuffer[9] = (byte)(number >>> 48);
+        writeBuffer[10] = (byte)(number >>> 40);
+        writeBuffer[11] = (byte)(number >>> 32);
+        writeBuffer[12] = (byte)(number >>> 24);
+        writeBuffer[13] = (byte)(number >>> 16);
+        writeBuffer[14] = (byte)(number >>>  8);
+        writeBuffer[15] = (byte)(number >>>  0);
+        
+        writeBuffer[16]=(byte) getSendRecord().size();
+        
+        writeBuffer[17]=(byte) (avaliable ? 1 : 0);
+		dto.write(writeBuffer);
+	}
+	 private final byte[] readBuffer = new byte[18];
 	@Override
 	protected void readFromStream(DataInput din) throws IOException {
 		super.readFromStream(din);
-		sport=din.readInt();
+		/*sport=din.readInt();
 		dport=din.readInt();
 		number=din.readLong();
-		avaliable=din.readBoolean();
+		sendcount=din.readUnsignedByte();
+		avaliable=din.readBoolean();*/
+		
+		din.readFully(readBuffer);
+		sport=((readBuffer[0] << 24) + (readBuffer[1] << 16) + (readBuffer[2] << 8) + (readBuffer[3] << 0));
+		dport=((readBuffer[4] << 24) + (readBuffer[5] << 16) + (readBuffer[6] << 8) + (readBuffer[7] << 0));
+		number=(((long)readBuffer[8] << 56) +
+                ((long)(readBuffer[9] & 255) << 48) +
+                ((long)(readBuffer[10] & 255) << 40) +
+                ((long)(readBuffer[11] & 255) << 32) +
+                ((long)(readBuffer[12] & 255) << 24) +
+                ((readBuffer[13] & 255) << 16) +
+                ((readBuffer[14] & 255) <<  8) +
+                ((readBuffer[15] & 255) <<  0));
+		sendcount=readBuffer[16]&0xff;
+		avaliable=(readBuffer[17] != 0);
 	}
+
 }

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

@@ -5,7 +5,7 @@ import java.io.DataOutput;
 import java.io.IOException;
 import java.net.Inet6Address;
 
-public class LINESPacket extends KLALBPacket {
+public class ADDLINESPacket extends KLALBPacket {
 	private String lines;
 
 
@@ -13,13 +13,13 @@ public class LINESPacket extends KLALBPacket {
 		return lines;
 	}
 
-	public LINESPacket(String lines) {
-		super(LINES);
+	public ADDLINESPacket(String lines) {
+		super(ADDLINES);
 		this.lines=lines;
 	}
 
-	public LINESPacket() {
-		super(LINES);
+	public ADDLINESPacket() {
+		super(ADDLINES);
 	}
 	private int UTFlength(String str) {
 		 int strlen = str.length();
@@ -38,7 +38,7 @@ public class LINESPacket extends KLALBPacket {
 	}
 	@Override
 	public String toString() {
-		return "LINES\n"+lines;
+		return "ADDLINES\n"+lines;
 	}
 
 	@Override

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

@@ -11,14 +11,14 @@ public class ByteArrayRecycle {
 		this.capacity = capacity;
 		this.length = length;
 	}
-	public synchronized void recycle(byte[]b) {
+	public void recycle(byte[]b) {
 		if(b.length!=length)
 			throw new IllegalArgumentException("wrong length");
 		if(rec.size()<capacity) {
 			rec.add(b);
 		}
 	}
-	public synchronized byte[] create() {
+	public byte[] create() {
 		byte[]b=rec.poll();
 		if(b==null) {
 			b=new byte[length];

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

@@ -11,7 +11,8 @@ import java.util.zip.GZIPInputStream;
 import java.util.zip.Inflater;
 import java.util.zip.InflaterInputStream;
 
-import org.kne.cloud.network.mport.SocketBridge;
+import org.kne.cloud.network.SocketBridge;
+import org.kne.cloud.network.StreamBridge;
 
 public class CompressedSocketBridge extends SocketBridge {
 
@@ -20,15 +21,10 @@ public class CompressedSocketBridge extends SocketBridge {
 	}
 
 	@Override
-	protected OutputStream getBOUT() throws IOException {
-		return new DeflaterOutputStream(super.getBOUT(), new Deflater(Deflater.BEST_COMPRESSION, true), true) ;
+	protected void createStreamBridge() throws IOException {
+		bridgeAB = new StreamBridge(a.getInputStream(),new DeflaterOutputStream( b.getOutputStream(), new Deflater(Deflater.BEST_COMPRESSION, true), true) );
+		bridgeBA = new StreamBridge(new InflaterInputStream( b.getInputStream(), new Inflater(true)) , a.getOutputStream());
 	}
 
-	@Override
-	protected InputStream getBIN() throws IOException {
-		return new InflaterInputStream(super.getBIN(), new Inflater(true)) ;
-	}
-
-	
 	
 }

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

@@ -2,19 +2,29 @@ package org.kne.cloud.network.klalb;
 
 import java.io.DataInput;
 import java.io.DataOutput;
+import java.io.DataOutputStream;
 import java.io.IOException;
+import java.io.StreamCorruptedException;
+import java.util.Objects;
+import java.util.zip.CRC32;
 
-public class DATATPacket extends KLALBPacket {
-	public static final ByteArrayRecycle arrayRecycle=new ByteArrayRecycle(100,65535);
+import javax.sound.sampled.Port;
+
+public class DATATPacket extends KLALBPacket implements Comparable<DATATPacket>,PortPacket{
+	public static final ByteArrayRecycle arrayRecycle=new ByteArrayRecycle(5000,65535);
 	@Override
 	public long getLength() {
-		return super.getLength()+18+size;
+		return super.getLength()+19+size;
 	}
 
 	private int sport,dport;
 	private long number;
 	private int size;
 	private byte[]data;
+	private int sendcount;
+
+	volatile long resendtimer=System.nanoTime();
+	
 	public DATATPacket(int sport,int dport,long number,byte[]data,int size) {
 		super(DATAT);
 		this.sport=sport;
@@ -24,6 +34,10 @@ public class DATATPacket extends KLALBPacket {
 		this.size=size;
 	}
 
+	public int getSendcount() {
+		return sendcount;
+	}
+
 	public DATATPacket() {
 		super(DATAT);
 	}
@@ -48,28 +62,107 @@ public class DATATPacket extends KLALBPacket {
 	public String toString() {
 		return "DATAT "+sport+"->"+dport+" "+number+"["+size+"]";
 	}
-
+	 private final byte[] writeBuffer = new byte[19];
 	@Override
 	protected void writeToStream(DataOutput dto) throws IOException {
 		super.writeToStream(dto);
-		dto.writeInt(sport);
+		/*dto.writeInt(sport);
 		dto.writeInt(dport);
 		dto.writeLong(number);
-		dto.writeChar(size);
+		dto.writeByte(getSendRecord().size());
+		dto.writeChar(size);*/
+		
+		writeBuffer[0] = (byte)(sport >>> 24);
+        writeBuffer[1] = (byte)(sport >>> 16);
+        writeBuffer[2] = (byte)(sport >>>  8);
+        writeBuffer[3] = (byte)(sport >>>  0);
+        
+    	writeBuffer[4] = (byte)(dport >>> 24);
+        writeBuffer[5] = (byte)(dport >>> 16);
+        writeBuffer[6] = (byte)(dport >>>  8);
+        writeBuffer[7] = (byte)(dport >>>  0);
+
+        writeBuffer[8] = (byte)(number >>> 56);
+        writeBuffer[9] = (byte)(number >>> 48);
+        writeBuffer[10] = (byte)(number >>> 40);
+        writeBuffer[11] = (byte)(number >>> 32);
+        writeBuffer[12] = (byte)(number >>> 24);
+        writeBuffer[13] = (byte)(number >>> 16);
+        writeBuffer[14] = (byte)(number >>>  8);
+        writeBuffer[15] = (byte)(number >>>  0);
+        
+        writeBuffer[16]=(byte) getSendRecord().size();
+        
+        writeBuffer[17]=(byte) (size>>>8);
+        writeBuffer[18]=(byte) (size>>>0);
+		dto.write(writeBuffer);
+        //CRC32 crc=new CRC32();
+       // crc.update(data, 0, size);
 		dto.write(data,0,size);
+		//dto.writeLong(crc.getValue());
 	}
 
+	 private final byte[] readBuffer = new byte[19];
 	@Override
 	protected void readFromStream(DataInput din) throws IOException {
 		super.readFromStream(din);
-		sport=din.readInt();
+		/*sport=din.readInt();
 		dport=din.readInt();
 		number=din.readLong();
-		size=din.readChar();
+		sendcount=din.readUnsignedByte();
+		size=din.readChar();*/
+		
+		din.readFully(readBuffer);
+		sport=((readBuffer[0] << 24) + (readBuffer[1] << 16) + (readBuffer[2] << 8) + (readBuffer[3] << 0));
+		dport=((readBuffer[4] << 24) + (readBuffer[5] << 16) + (readBuffer[6] << 8) + (readBuffer[7] << 0));
+		number=(((long)readBuffer[8] << 56) +
+                ((long)(readBuffer[9] & 255) << 48) +
+                ((long)(readBuffer[10] & 255) << 40) +
+                ((long)(readBuffer[11] & 255) << 32) +
+                ((long)(readBuffer[12] & 255) << 24) +
+                ((readBuffer[13] & 255) << 16) +
+                ((readBuffer[14] & 255) <<  8) +
+                ((readBuffer[15] & 255) <<  0));
+		sendcount=readBuffer[16]&0xff;
+		size=(((readBuffer[17]&0xff) << 8) + ((readBuffer[18]&0xff) << 0));
+		
 		data=arrayRecycle.create();
 		din.readFully(data,0,size);
+		/*CRC32 crc32=new CRC32();
+		crc32.update(data, 0, size);
+		if(din.readLong()!=crc32.getValue()) {
+			throw new StreamCorruptedException("CRC32 error!");
+		}*/
 	}
+	@Override
+	public int hashCode() {
+		return Objects.hash(number);
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (this == obj)
+			return true;
+		if (obj == null)
+			return false;
+		if (getClass() != obj.getClass())
+			return false;
+		DATATPacket other = (DATATPacket) obj;
+		return number == other.number;
+	}
+
 	public int getSize() {
 		return size;
 	}
+
+	@Override
+	public int compareTo(DATATPacket o) {
+		if(number>o.number) {
+			return 1;
+		}else if(number<o.number){
+			return -1;
+		}else {
+		return 0;
+		}
+	}
 }

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

@@ -6,12 +6,15 @@ import java.net.ConnectException;
 import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.net.NetworkInterface;
 import java.net.NoRouteToHostException;
 import java.net.Socket;
 import java.net.SocketAddress;
+import java.net.SocketException;
 import java.net.SocketTimeoutException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Enumeration;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -24,12 +27,22 @@ import java.util.UUID;
 import java.util.Vector;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.BiConsumer;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 
-import org.kne.cloud.network.mport.MultipurposeSocketAddress;
-import org.kne.cloud.network.mport.NetworkService;
-import org.kne.cloud.network.mport.ProxyProfileEntry;
-import org.kne.cloud.network.mport.ThreadTool;
+import javax.management.openmbean.ArrayType;
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+
+import org.kne.cloud.network.MultipurposeSocketAddress;
+import org.kne.cloud.network.NetworkService;
+import org.kne.cloud.network.Proxy;
+import org.kne.cloud.network.SocketType;
+import org.kne.cloud.network.ThreadTool;
+import org.kne.debug.TimeDebugger;
 
 import javassist.ClassPool;
 import javassist.CtClass;
@@ -39,12 +52,21 @@ import javassist.bytecode.CodeAttribute;
 import javassist.bytecode.CodeIterator;
 
 public class KLALBController {
-	private LineManager lineManager;
+	private SocketType socketType= new KLALBSocketType();
+	public class KLALBSocketType extends SocketType{
+
+		public KLALBSocketType() {
+			super(new KLALBVirtualSocketFactory(KLALBController.this), new KLALBVirtualServerSocketFactory(KLALBController.this));
+		}
+		
+	}
+	public SocketType getSocketType() {
+		return socketType;
+	}
+
 	private Supplier<String> selflineTableSupplier=()->{return null;};
 	
 
-
-
 	public Supplier<String> getSelflineTableSupplier() {
 		return selflineTableSupplier;
 	}
@@ -53,150 +75,157 @@ public class KLALBController {
 		this.selflineTableSupplier = selflineTableSupplier;
 	}
 
-	public LineManager getLineManager() {
-		if(lineManager==null)
-			lineManager=createLineManager();
-		return lineManager;
-	}
-
-	protected LineManager createLineManager() {
-		return new LineManager(this);
-	}
-
 	private Inet6Address self;
 
 	public Inet6Address getSelf() {
 		return self;
 	}
+	private List<KLALBRemoteLine> lines=new ArrayList<>();
+	//private ReadWriteLock lineslock=new ReentrantReadWriteLock();
 	
-	private Map<Inet6Address, List<KLALBRemoteSocket>> routes = new ConcurrentHashMap<>();
-
-	protected Map<Inet6Address, List<KLALBRemoteSocket>> getRoutes() {
-		return routes;
+	public List<KLALBRemoteLine> getLines() {
+		return lines;
 	}
 
-	private void setLine(Inet6Address vaddr, KLALBRemoteSocket krs) {
-		synchronized (routes) {
-			List<KLALBRemoteSocket> al = routes.computeIfAbsent(vaddr, (vaddr2) -> {
-				return new ArrayList<KLALBRemoteSocket>();
-			});
-			al.add(krs);
-		}
+	private PortBinder streamPortBinder=new PortBinder(this);
+	
+	protected PortBinder getStreamPortBinder() {
+		return streamPortBinder;
 	}
 
-	private void removeLine(KLALBRemoteSocket krs) {
-		synchronized (routes) {
-			Iterator<Entry<Inet6Address, List<KLALBRemoteSocket>>> iter = routes.entrySet().iterator();
-			while (iter.hasNext()) {
-				Map.Entry<java.net.Inet6Address, java.util.List<org.kne.cloud.network.klalb.KLALBRemoteSocket>> entry = (Map.Entry<java.net.Inet6Address, java.util.List<org.kne.cloud.network.klalb.KLALBRemoteSocket>>) iter
-						.next();
-				entry.getValue().remove(krs);
-				if (entry.getValue().isEmpty()) {
-					iter.remove();
-				}
+
+	public void reconnectImmediately() {
+		synchronized (lines) {
+			for (Iterator<KLALBRemoteLine> iterator = lines.iterator(); iterator.hasNext();) {
+				KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
+				klalbRemoteLine.reconnectImmediately();
 			}
 		}
 	}
 	
-	public void addRemoteSocket(KLALBRemoteSocket krs) {
-		krs.setController(this);
-		CountDownLatch cdl = new CountDownLatch(1);
-		krs.setPacketReceiver((rec) -> {
+	private PacketReceiver prc=new PacketReceiver();
+	private class PacketReceiver implements KLALBPacketConsumer{
+
+		@Override
+		public void accept(KLALBRemoteLine krs, KLALBPacket rec) {
 			try {
-				if (rec instanceof SYNTPacket) {
-					SYNTPacket synt = (SYNTPacket) rec;
-					KLALBVirtualSocketImpl kvi = bindmap.get(synt.getDport());
-					if (kvi != null) {
-						if (kvi.isListening()) {
-							kvi.getPackReceiver().accept(krs.getRemoteVaddr(), synt);
+				if(rec instanceof PortPacket&&krs.getRemoteVaddr()!=null) {
+					PortPacket pt=(PortPacket) rec;
+					if(!streamPortBinder.distributePacketToConsumer(krs, pt)) {
+						if(!(pt instanceof RSTPacket))
+						sendPacketToAddress(krs.getRemoteVaddr(), new RSTPacket(pt.getDport(), pt.getSport()),
+								0,2);
 					}
-					} else {
-
-						sendPacketToAddress(krs.getRemoteVaddr(), new RSTPacket(synt.getDport(), synt.getSport()),
-								65537);
-					}
-				} else if (rec instanceof SACKTPacket) {
-					SACKTPacket sackt = (SACKTPacket) rec;
-					KLALBVirtualSocketImpl kvi = bindmap.get(sackt.getDport());
-					if (kvi != null) {
-						if (!kvi.isListening()) {
-							kvi.getPackReceiver().accept(krs.getRemoteVaddr(), sackt);
-						}
-					}
-				} else if (rec instanceof RSTPacket) {
-					RSTPacket rst = (RSTPacket) rec;
-					KLALBVirtualSocketImpl kvi = bindmap.get(rst.getDport());
-					if (kvi != null) {
-						if (kvi.isListening()) {
-							KLALBVirtualSocketImpl kvi2 = kvi.getAccepts()
-									.get(new InetSocketAddress(krs.getRemoteVaddr(), rst.getSport()));
-							if (kvi2 != null) {
-								kvi2.getPackReceiver().accept(krs.getRemoteVaddr(), rst);
-							}
-						} else {
-							kvi.getPackReceiver().accept(krs.getRemoteVaddr(), rst);
-						}
-					}
-				} else if (rec instanceof DATATPacket) {
-					DATATPacket datat = (DATATPacket) rec;
-					KLALBVirtualSocketImpl kvi = bindmap.get(datat.getDport());
-					if (kvi != null) {
-						if (kvi.isListening()) {
-							KLALBVirtualSocketImpl kvi2 = kvi.getAccepts()
-									.get(new InetSocketAddress(krs.getRemoteVaddr(), datat.getSport()));
-							if (kvi2 != null) {
-								kvi2.getPackReceiver().accept(krs.getRemoteVaddr(), datat);
-							}
-						} else {
-							kvi.getPackReceiver().accept(krs.getRemoteVaddr(), datat);
-						}
-					}
-				} else if (rec instanceof ACKTPacket) {
-					ACKTPacket ackt = (ACKTPacket) rec;
-					KLALBVirtualSocketImpl kvi = bindmap.get(ackt.getDport());
-					if (kvi != null) {
-						if (kvi.isListening()) {
-							KLALBVirtualSocketImpl kvi2 = kvi.getAccepts()
-									.get(new InetSocketAddress(krs.getRemoteVaddr(), ackt.getSport()));
-							if (kvi2 != null) {
-								kvi2.getPackReceiver().accept(krs.getRemoteVaddr(), ackt);
-							}
-						} else {
-							kvi.getPackReceiver().accept(krs.getRemoteVaddr(), ackt);
-						}
-					}
-				} else if (rec instanceof VADDRPacket) {
-					VADDRPacket var = (VADDRPacket) rec;
-					setLine(var.getVaddr(), krs);
-					cdl.countDown();
-				}else if(rec instanceof LINESPacket) {
-					LINESPacket lpt=(LINESPacket) rec;
+				}else {
+				switch (rec.getType()) {
+				case KLALBPacket.ADDLINES:
+					ADDLINESPacket lpt=(ADDLINESPacket) rec;
 					String s=lpt.getLines();
 					Scanner scn=new Scanner(s);
 					while(scn.hasNext()) {
 						String sn=scn.nextLine();
-						getLineManager().addHostPort(new MultipurposeSocketAddress(sn));
+						MultipurposeSocketAddress msa=	new MultipurposeSocketAddress(sn);
+						addRemoteLines(msa);
+						
+					}
+					break;
+				default:
+					break;
 				}
 				}
-			} catch (NoRouteToHostException e) {
+			} catch (IOException e) {
+				// TODO 自动生成的 catch 块
 				e.printStackTrace();
 			}
-		});
-		krs.setCloseListener((x) -> {
-			removeLine(krs);
-			cdl.countDown();
-		});
-		krs.sendPacket(new VADDRPacket(self), 65537);
+		}
+
+		
+	}
+	public void addRemoteLines(MultipurposeSocketAddress target) throws SocketTimeoutException, SocketException {
+		synchronized (lines) {
+			
+				try {
+					Enumeration<NetworkInterface>eu= NetworkInterface.getNetworkInterfaces();
+					while (eu.hasMoreElements()) {
+						NetworkInterface networkInterface = (NetworkInterface) eu.nextElement();
+						if(networkInterface.isUp()) {
+						//System.out.println(networkInterface+" "+networkInterface.isUp());
+						Enumeration<InetAddress>ei= networkInterface.getInetAddresses();
+						while (ei.hasMoreElements()) {
+							InetAddress inetAddress = (InetAddress) ei.nextElement();
+							MultipurposeSocketAddress bind=new MultipurposeSocketAddress(inetAddress.getHostAddress(),0);
+							if(!checkContainsTargetAndBind(target,bind)) {
+								//System.out.println(target+" "+bind);
+							addRemoteLine( new KLALBRemoteLine(target,bind));
+							}
+						}
+						}
+					}
+				} catch (SocketException e) {
+					if(!checkContainsTarget(target))
+					addRemoteLine( new KLALBRemoteLine(target));
+					throw e;
+				}
+			
+		}
+	}
+	public Inet6Address getRemoteVaddrBySocketAddress(MultipurposeSocketAddress target) throws SocketTimeoutException {
+		KLALBRemoteLine kr=null;
+		synchronized(lines) {
+		for (Iterator iterator = lines.iterator(); iterator.hasNext();) {
+			KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
+			if(target.equals(klalbRemoteLine.getSocketAddress())) {
+				kr=klalbRemoteLine;
+			}
+		}
+		}
+		if(kr==null) {
+		kr=new KLALBRemoteLine(target);
+	addRemoteLine(kr);
+	kr.waitForRemoteVaddrAvaliable(20000);
+		}else {
+			kr.reconnectImmediately();
+			kr.waitForRemoteVaddrAvaliable(20000);
+		}
+		return kr.getRemoteVaddr();
+	}
+	private boolean checkContainsTargetAndBind(MultipurposeSocketAddress target,MultipurposeSocketAddress bind) {
+		boolean b=false;
+		for (Iterator<KLALBRemoteLine> iterator = lines.iterator(); iterator.hasNext();) {
+			KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
+			if(bind.equals(klalbRemoteLine.getBindAddress())&&target.equals(klalbRemoteLine.getSocketAddress())) {
+				b=true;
+				break;
+			}
+		}
+		return b;
+	}
+
+	private boolean checkContainsTarget(MultipurposeSocketAddress target) {
+		boolean b=false;
+		for (Iterator<KLALBRemoteLine> iterator = lines.iterator(); iterator.hasNext();) {
+			KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
+			if(target.equals(klalbRemoteLine.getSocketAddress())) {
+				b=true;
+				break;
+			}
+		}
+		return b;
+	}
+
+	public void addRemoteLine(KLALBRemoteLine krs) throws SocketTimeoutException {
+		krs.setPacketReceiver(prc);
+		krs.setLocalVaddrSupplier(()->{return self;});
+		krs.startIO();
 		String selflineTable=selflineTableSupplier.get();
 		if(selflineTable!=null)
-		krs.sendPacket(new LINESPacket(selflineTable), 65537);
-		try {
-			cdl.await();
-		} catch (InterruptedException e) {
-			e.printStackTrace();
+		krs.sendPacket(new ADDLINESPacket(selflineTable), 0);
+		synchronized (lines) {
+		lines.add(krs);
 		}
 	}
 
+
 	public KLALBController(Inet6Address self) {
 		this.self = self;
 	}
@@ -205,113 +234,127 @@ public class KLALBController {
 		this.self=KLALBUtils.uuidToIP(UUID.randomUUID());
 	}
 
-	private Map<Integer, KLALBVirtualSocketImpl> bindmap = new ConcurrentHashMap<>();
-
-	protected Map<Integer, KLALBVirtualSocketImpl> getBindmap() {
-		return bindmap;
-	}
-
 	protected KLALBVirtualSocketImpl createVirtualImpl() {
 		return new KLALBVirtualSocketImpl(this);
 	}
 
-	protected int bind(KLALBVirtualSocketImpl klalbVirtualSocketImpl, int port) throws BindException {
-		synchronized (bindmap) {
-			if (port == 0) {
-				port = allocPort();
-			}
-			if (bindmap.putIfAbsent(port, klalbVirtualSocketImpl) != null) {
-				throw new BindException("port " + port + " is already bind!");
-			}
-			return port;
-		}
-	}
 	
-	protected void unbind(KLALBVirtualSocketImpl klalbVirtualSocketImpl) {
-		synchronized (bindmap) {
-			Set<Entry<Integer, KLALBVirtualSocketImpl>> s = bindmap.entrySet();
-			Iterator<Entry<Integer, KLALBVirtualSocketImpl>> it = s.iterator();
-			while (it.hasNext()) {
-				Entry<Integer, KLALBVirtualSocketImpl> object = it.next();
-				if (klalbVirtualSocketImpl.equals(object.getValue())) {
-					it.remove();
-					return;
-				}
-			}
-		}
-	}
-
-	protected int allocPort() throws BindException {
-		int i = 1;
-		while (bindmap.containsKey(i)) {
-			if (i == 65535) {
-				throw new BindException("can't alloc port");
-			}
-			i++;
-		}
-		return i;
-	}
 
 	protected void sendPacketToAddress(Inet6Address addr, KLALBPacket syntPacket, int priority)
-			throws NoRouteToHostException {
+			throws IOException {
 		sendPacketToAddress(addr, syntPacket, priority, 1);
 	}
 	
+			private void updateLines2(Inet6Address addr) throws SocketTimeoutException {
+				List<KLALBRemoteLine> l=new ArrayList();
+				synchronized (lines) {
+				for (int i = 0; i < lines.size(); i++) {
+					KLALBRemoteLine klalbRemoteLine = lines.get(i);
+					if(klalbRemoteLine.isClosed()) {
+						lines.remove(i);
+						i--;
+						continue;
+					}
+					if(addr.equals(klalbRemoteLine.getRemoteVaddr())&&klalbRemoteLine.getMonitor().getState()==Monitor.ONLINE) {
+						l.add(klalbRemoteLine);
+					}
+				}
+		}
+				if(l.isEmpty()) {
+					lines2.remove(addr);
+				}else {
+				lines2.put(addr, l);
+				}
+			}
+			private	Map<Inet6Address,List<KLALBRemoteLine>> lines2 = new ConcurrentHashMap<>();
+			private volatile long itm=System.nanoTime();
 	protected void sendPacketToAddress(Inet6Address addr, KLALBPacket packet, int priority, int count)
-			throws NoRouteToHostException {
-		synchronized (routes) {
-			List<KLALBRemoteSocket> l = routes.get(addr);
-			if (l == null || l.isEmpty()) {
+			throws IOException {
+		/*if(packet instanceof RSTPacket) {
+			new Exception("-RST-").printStackTrace();
+		}*/
+		//TimeDebugger tdb=new TimeDebugger();
+		//tdb.putTime("start");
+		
+		List<KLALBRemoteLine> lines2x;
+			//loop:while(true) {
+				long cur=System.nanoTime();
+			if(cur-itm>10000000L) {
+				itm=cur;
+				lines2.clear();
+			}
+				lines2x=lines2.get(addr);
+				if (lines2x == null || lines2x.isEmpty()) {
+					updateLines2(addr);
+					lines2x=lines2.get(addr);
+				}
+				if (lines2x == null || lines2x.isEmpty()) {
 					throw new NoRouteToHostException("address unreachable: " + addr);
 				}
-			List<KLALBRemoteSocket> l2 = (List<KLALBRemoteSocket>) ((ArrayList<KLALBRemoteSocket>) l).clone();
+			//tdb.putTime("selectLines");
+			/*	for (Iterator<KLALBRemoteLine> iterator = lines2x.iterator(); iterator.hasNext();) {
+					KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
+					if(klalbRemoteLine.statLengthBefore(priority)<=65536*10) {
+						break loop;
+					}
+				}
+				try {
+					//System.out.println("slp");
+					Thread.sleep(1);
+				} catch (InterruptedException e) {
+					e.printStackTrace();
+				}
+			}*/
+			List<KLALBRemoteLine> l2 = (List<KLALBRemoteLine>) ((ArrayList<KLALBRemoteLine>) lines2x).clone();
 			l2.removeAll(packet.getSendRecord());
 			if(l2.isEmpty()) {
-				l2 = (List<KLALBRemoteSocket>) ((ArrayList<KLALBRemoteSocket>) l).clone();
+				l2 = (List<KLALBRemoteLine>) ((ArrayList<KLALBRemoteLine>) lines2x).clone();
 			}
+
+			//tdb.putTime("findAvaliable");
+			
 			int count0 = Math.min(count, l2.size());
-			LineDecitionComparator ldc = new LineDecitionComparator(l2,packet, priority);
-			Collections.sort(l2,ldc);
+			l2.forEach((r)->{
+				r.runPredict(packet,priority);
+			});
+			//Collections.shuffle(l2);
+			Collections.sort(l2);
+			//tdb.putTime("makeDecision");
 			//System.out.println(l2);
-			for (Iterator<KLALBRemoteSocket> iterator = l2.iterator(); iterator.hasNext();) {
-				KLALBRemoteSocket krst = (KLALBRemoteSocket) iterator.next();
+			for (int i = 0; i < l2.size(); i++) {
+				KLALBRemoteLine krst =l2.get(i);
 				krst.sendPacket(packet, priority);
 				packet.getSendRecord().add(krst);
 				count0--;
-				Thread.yield();
 				if (count0 <= 0)
 					break;
 			}
-		}
-
+			//tdb.putTime("sendPacket");
+			//tdb.print();
+			/*if(packet instanceof RSTPacket)
+			new Exception().printStackTrace();*/
 	}
 	protected void removeFromSend(Inet6Address addr,KLALBPacket klalbPacket) {
-		synchronized (routes) {
-			List<KLALBRemoteSocket> l = routes.get(addr);
-			if (l != null && !l.isEmpty()) {
-				l.forEach((x)->{
-					x.remoeFromSendQueue(klalbPacket);
-				});
+		synchronized (lines) {
+			for (Iterator<KLALBRemoteLine> iterator = lines.iterator(); iterator.hasNext();) {
+				KLALBRemoteLine klalbRemoteLine = (KLALBRemoteLine) iterator.next();
+				klalbRemoteLine.remoeFromSendQueue(klalbPacket);
 			}
 		}
 	}
-	protected boolean checkIsBind(KLALBVirtualSocketImpl klalbVirtualSocketImpl) {
-		return bindmap.containsValue(klalbVirtualSocketImpl);
-	}
 private Timer t=new Timer("数据包发送计时器", true);
 	public Timer getTimer() {
 		return t;
 	}
 	
+	
 	public void registerToProxyTypeAs(String proxyname) {
-		MultipurposeSocketAddress.getSocketFactoryRegister().put(proxyname, new KLALBVirtualSocketFactory(this));
-		MultipurposeSocketAddress.getServerSocketFactoryRegister().put(proxyname, new KLALBVirtualServerSocketFactory(this));
-		ProxyProfileEntry.getRegister().put(proxyname, new KLALBNetworkService());
+		MultipurposeSocketAddress.getSocketTypeRegister().put(proxyname+"_Stream",socketType);
+		//ProxyProfileEntry.getRegister().put(proxyname, this);
 	}
 
-	private class KLALBNetworkService implements NetworkService{
 
-		@Override
+	/*@Override
 	public void listen(MultipurposeSocketAddress msa) {
 		// TODO 自动生成的方法存根
 		
@@ -325,15 +368,15 @@ private Timer t=new Timer("数据包发送计时器", true);
 
 	@Override
 	public void connect(MultipurposeSocketAddress msa) {
-			getLineManager().addHostPort(msa);
+		// TODO 自动生成的方法存根
+		
 	}
 
 	@Override
 	public void unconnect(MultipurposeSocketAddress msc) {
-			getLineManager().removeHostPort(msc);
-		}
+		// TODO 自动生成的方法存根
 		
-	}
+	}*/
 
 	
 }

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

@@ -1,5 +1,7 @@
 package org.kne.cloud.network.klalb;
 import static org.kne.cloud.network.klalb.KLALBPacket.*;
+
+import java.io.DataInput;
 import java.io.DataInputStream;
 import java.io.EOFException;
 import java.io.IOException;
@@ -21,48 +23,67 @@ public class KLALBInputStream extends DataInputStream {
 		if(bv!=2)
 			throw new StreamCorruptedException("remote version is V"+bv+"."+sv+",not V2.0");
 	}
-	public synchronized KLALBPacket readPacket() throws IOException {
-		int type=read();
+	public KLALBPacket readPacket() throws IOException {
+		return readKLALBPacketFromStream(this);
+	}
+	public static KLALBPacket readKLALBPacketFromStream(DataInputStream in) throws IOException {
+int type=in.read();
 		
 		if(type==-1) {
-			throw new EOFException();
+			return null;
 		}
 		KLALBPacket klp;
 		switch(type) {
 		case PING:klp=new PINGPacket();
-			klp.readFromStream(this);
+			klp.readFromStream(in);
 			return klp;
 		case PONG:
 			klp=new PONGPacket();
-			klp.readFromStream(this);
+			klp.readFromStream(in);
 			return klp;
 		case SYNT:
 			klp=new SYNTPacket();
-			klp.readFromStream(this);
+			klp.readFromStream(in);
 			return klp;
 		case SACKT:
 			klp=new SACKTPacket();
-			klp.readFromStream(this);
+			klp.readFromStream(in);
 			return klp;
 		case RST:
 			klp=new RSTPacket();
-			klp.readFromStream(this);
+			klp.readFromStream(in);
 			return klp;
 		case DATAT:
 			klp=new DATATPacket();
-			klp.readFromStream(this);
+			klp.readFromStream(in);
 			return klp;
 		case ACKT:
 			klp=new ACKTPacket();
-			klp.readFromStream(this);
+			klp.readFromStream(in);
 			return klp;
 		case VADDR:
 			klp=new VADDRPacket();
-			klp.readFromStream(this);
+			klp.readFromStream(in);
 			return klp;
-		case LINES:
-			klp=new LINESPacket();
-			klp.readFromStream(this);
+		case ADDLINES:
+			klp=new ADDLINESPacket();
+			klp.readFromStream(in);
+			return klp;
+		case NACKT:
+			klp=new NACKTPacket();
+			klp.readFromStream(in);
+			return klp;	
+			case TEST:
+				klp=new TESTPacket();
+				klp.readFromStream(in);
+				return klp;
+			case VADDRACK:
+				klp=new VADDRACKPacket();
+				klp.readFromStream(in);
+				return klp;
+			case VADDRREQ:
+				klp=new VADDRREQPacket();
+				klp.readFromStream(in);
 				return klp;
 		}
 		throw new StreamCorruptedException("unknown package type:"+type);

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

@@ -1,72 +1,58 @@
 package org.kne.cloud.network.klalb;
 
+import java.io.BufferedReader;
 import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
 import java.io.IOException;
 import java.util.Iterator;
-import java.util.List;
 import java.util.Scanner;
 
-import org.kne.cloud.network.klalb.LineManager.LineEntry;
-import org.kne.cloud.network.mport.MultipurposeSocketAddress;
-import org.kne.cloud.network.mport.ProxyProfileAnalyser;
-import org.kne.cloud.network.mport.ProxyProfileExecutor;
-import org.kne.cloud.network.mport.TCPListener;
+import org.kne.cloud.network.KLALBProxyConfigJsonExecuter;
+import org.kne.cloud.network.SocketToServiceProxy;
+import org.kne.cloud.network.SocketToSocketProxy;
 
 public class KLALBMain {
-	public static TCPListener tcpl;
-	public static ServerPropties sp;
-	public static KLALBController kc;
 
-	public static ProxyProfileExecutor pfa;
-	public static File f=new File("lines.cfg");
 	public static void main(String[] args) throws IOException {
-		
-		System.out.println("KLALB负载均衡V2.0");
-		sp=new ServerPropties();
-		
-		System.out.println("虚拟地址："+sp.getVirtualIP().getHostAddress());
-		kc=new KLALBController(sp.getVirtualIP());
-		kc.registerToProxyTypeAs("KLALB");
-	
-		
-		
-		
-		pfa=new ProxyProfileExecutor();
-		pfa.load(f);
-		
+		System.out.println(CONST.klalb+" V"+CONST.klalbver);
 		Scanner scn=new Scanner(System.in);
+		
+		File configJson=new File("klalbconfig.json");
+		
+		KLALBProxyConfigJsonExecuter kpcje=new KLALBProxyConfigJsonExecuter();
+		kpcje.loadConfigJson(configJson);
 		while(true) {
 		String s=scn.next();
 		String[]sc=s.split(" ");
 		switch(sc[0]) {
 		case "help":
+			System.out.print("help:查看命令使用说明");
 			System.out.print("state:查看线路状态");
-			System.out.println("reload:重新加载线路配置");
-		break;
-		case "reloadlines":
-			pfa.load(f);
-			System.out.println("重新加载线路配置成功");
+			//System.out.println("reload:重新加载线路配置文件");
+			System.out.println("reconnect:所有离线线路跳过重连等待时间立即尝试重连");
+			System.out.println("stop:退出程序");
+			
 		break;
 		case "state":
-			LineManager le=kc.getLineManager();
-			for (Iterator<java.util.Map.Entry<MultipurposeSocketAddress, LineEntry>> iterator = le.entrySet().iterator(); iterator.hasNext();) {
-				java.util.Map.Entry<MultipurposeSocketAddress, LineEntry> hostPort =  iterator.next();
-				System.out.println(hostPort.getValue() .toString());
+			System.out.println("状态\t上传流量\t下载流量\t上传速度\t下载速度\t延迟\t抖动\t下一次重试");
+			for (Iterator<KLALBRemoteLine> iterator = kpcje.getKlalbController().getLines().iterator(); iterator.hasNext();) {
+				KLALBRemoteLine hostPort =  iterator.next();
+				System.out.println(hostPort .toString2());
+				//System.out.println();
 			}
 			break;
+		case "stop":
+			System.exit(0);
+			break;
+		case "reconnect":
+			kpcje.getKlalbController().reconnectImmediately();
+			break;
 		default:
 			System.out.println("未知命令，请输入help以查询指令说明");
 		}
 		}
+		
 	}
-	private static void openPort(int port) throws IOException {
-		if(tcpl!=null)
-			tcpl.close();
-		tcpl=new TCPListener(new MultipurposeSocketAddress("0.0.0.0", port));
-		tcpl.setCon((soc)->{
-			KLALBRemoteSocket krs=new KLALBRemoteSocket(soc);
-			kc.addRemoteSocket(krs);
-		});
-		tcpl.open();
-	}
+
 }

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

@@ -12,11 +12,15 @@ public class KLALBOutputStream extends DataOutputStream {
 		byte[]b=new byte[] {'K','L','A','L','B'};
 		write(b);
 		writeInt(2);
-		writeInt(0);
+		writeInt(1);
 		flush();
 	}
-	public synchronized void writePacket(KLALBPacket klb) throws IOException {
-		write(klb.getType());
-		klb.writeToStream(this);
+	public void writePacket(KLALBPacket klb) throws IOException {
+		writeKLALBPacketToStream(this, klb);
+	}
+	public static void writeKLALBPacketToStream(DataOutputStream out,KLALBPacket klb) throws IOException {
+		out.write(klb.getType());
+		klb.writeToStream(out);
+		out.flush();
 	}
 }

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

@@ -8,7 +8,7 @@ import java.io.ObjectInput;
 import java.io.ObjectOutput;
 import java.util.Vector;
 
-public abstract class KLALBPacket{
+public abstract class KLALBPacket implements Sumable{
 	public static final int PING=0;
 	public static final int PONG=1;
 	public static final int SYNT=2;
@@ -18,7 +18,11 @@ public abstract class KLALBPacket{
 	public static final int ACKT=6;
 	public static final int DATAU=7;
 	public static final int VADDR=8;
-	public static final int LINES=9;
+	public static final int ADDLINES=9;
+	public static final int NACKT=10;
+	public static final int TEST=11;
+	public static final int VADDRACK=12;
+	public static final int VADDRREQ=13;
 	private int type;
 
 	public KLALBPacket(int type) {
@@ -32,18 +36,32 @@ public abstract class KLALBPacket{
 	public int getType() {
 		return type;
 	}
-	protected void writeToStream(DataOutput dto) throws IOException {
 	
+	private long sndtime,rcvtime;
+	protected void writeToStream(DataOutput dto) throws IOException {
+		sndtime=System.nanoTime();
 	}
 	protected void readFromStream(DataInput din) throws IOException {
-		
+		rcvtime=System.nanoTime();
+	}
+	public long getSndtime() {
+		return sndtime;
+	}
+	public long getRcvtime() {
+		return rcvtime;
 	}
 	public long getLength() {
 		return 1;
 	}
-	private Vector<KLALBRemoteSocket> sendRecord=new Vector<>();
+	@Override
+	public long getValue() {
+		return getLength();
+	}
 
-	public Vector<KLALBRemoteSocket> getSendRecord() {
+	
+	private Vector<KLALBRemoteLine> sendRecord=new Vector<>();
+
+	public Vector<KLALBRemoteLine> getSendRecord() {
 		return sendRecord;
 	}
 

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

@@ -5,6 +5,7 @@ import java.io.DataOutputStream;
 import java.io.IOException;
 import java.net.Inet6Address;
 import java.net.UnknownHostException;
+import java.util.List;
 import java.util.UUID;
 
 public class KLALBUtils {
@@ -54,4 +55,24 @@ public class KLALBUtils {
 	}
 		return fmt;
 	}
+	public static int binarySearchDATATPacketNumber(List<DATATPacket>lst,long number) {
+		int low =0;
+		int high=lst.size()-1;
+		int middle=0;
+		if(low>high||number<lst.get(low).getNumber()||number>lst.get(high).getNumber()) {
+			return -1;
+		}
+		while(low<=high) {
+			middle=(low+high)/2;
+			long xn=lst.get(middle).getNumber();
+			if(xn>number) {
+				high=middle-1;
+			}else if(xn<number) {
+				low=middle+1;
+			}else {
+				return middle;
+			}
+		}
+		return -1;
+	}
 }

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

@@ -7,8 +7,8 @@ import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketImpl;
 
-import org.kne.cloud.network.mport.VirtualServerSocket;
-import org.kne.cloud.network.mport.VirtualSocket;
+import org.kne.cloud.network.VirtualServerSocket;
+import org.kne.cloud.network.VirtualSocket;
 
 public class KLALBVirtualServerSocket extends VirtualServerSocket {
 	private KLALBController controler;

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

@@ -7,6 +7,7 @@ import java.net.ServerSocket;
 import javax.net.ServerSocketFactory;
 
 public class KLALBVirtualServerSocketFactory extends ServerSocketFactory {
+
 	private KLALBController controller;
 	
 	public KLALBVirtualServerSocketFactory(KLALBController controller) {
@@ -17,6 +18,10 @@ public class KLALBVirtualServerSocketFactory extends ServerSocketFactory {
 	public KLALBController getController() {
 		return controller;
 	}
+	@Override
+	public ServerSocket createServerSocket() throws IOException {
+		return new KLALBVirtualServerSocket(controller);
+	}
 
 	@Override
 	public ServerSocket createServerSocket(int port) throws IOException {

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

@@ -3,7 +3,7 @@ package org.kne.cloud.network.klalb;
 import java.io.IOException;
 import java.net.*;
 
-import org.kne.cloud.network.mport.VirtualSocket;
+import org.kne.cloud.network.VirtualSocket;
 
 public class KLALBVirtualSocket extends VirtualSocket {
 
@@ -59,4 +59,16 @@ public class KLALBVirtualSocket extends VirtualSocket {
              (SocketAddress) null);
     }
 
+	public void setCompress(int i) {
+		((KLALBVirtualSocketImpl)getVirtualImpl()).setCompress(i);
+	}
+	public int setCompress() {
+		return ((KLALBVirtualSocketImpl)getVirtualImpl()).getCompress();
+	}
+
+
+	public void associateSocket(Socket associateSocket) throws IOException {
+		((KLALBVirtualSocketImpl)getVirtualImpl()).associateSocket(associateSocket);
+	}
+	
 }

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

@@ -15,6 +15,10 @@ private KLALBController controller;
 		this.controller = controller;
 	}
 	@Override
+	public Socket createSocket() throws IOException {
+		return new KLALBVirtualSocket(controller);
+	}
+	@Override
 	public Socket createSocket(String arg0, int arg1) throws IOException, UnknownHostException {
 	
 		return new KLALBVirtualSocket(controller, arg0, arg1);

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

@@ -1,5 +1,6 @@
 package org.kne.cloud.network.klalb;
 
+import java.net.SocketException;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -7,28 +8,23 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
 
-public class LineDecitionComparator implements Comparator<KLALBRemoteSocket> {
-	private Map<KLALBRemoteSocket,Double> predictedlatencys=new HashMap<>();
-	public LineDecitionComparator (List<KLALBRemoteSocket> krs,KLALBPacket curr,int priority) {
-		for (Iterator<KLALBRemoteSocket> iterator = krs.iterator(); iterator.hasNext();) {
-			KLALBRemoteSocket klalbRemoteSocket = (KLALBRemoteSocket) iterator.next();
-			double x=klalbRemoteSocket.getMonitor().getLatency()/2;
-			 x+=klalbRemoteSocket.getMonitor().getJitter()/2;
-			AtomicLong al=new AtomicLong(0);
-			klalbRemoteSocket.getSendQueue().forEach((v)->{
-				if(v.getPriority()>=priority) {
-					al.addAndGet(v.getPacket().getLength());
-				}
-			});
-			al.addAndGet(curr.getLength());
+public class LineDecitionComparator implements Comparator<KLALBRemoteLine> {
+	private Map<KLALBRemoteLine,Long> predictedlatencys=new HashMap<>();
+	public LineDecitionComparator (List<KLALBRemoteLine> krs,KLALBPacket curr,int priority) {
+		for (Iterator<KLALBRemoteLine> iterator = krs.iterator(); iterator.hasNext();) {
+			KLALBRemoteLine klalbRemoteSocket = (KLALBRemoteLine) iterator.next();
+			long x=klalbRemoteSocket.getSndDelayFactor();
+			//long x=klalbRemoteSocket.getMonitor().getLatency()>>1;
+			/*long al=0;
+			al=klalbRemoteSocket.statLengthBefore(priority)+curr.getLength();
 			long speed=klalbRemoteSocket.getMonitor(). getOutSpeed();
 			if(speed==0) {
-				if(al.get()>0) {
+				if(al>0) {
 					x=Long.MAX_VALUE;
 				}
 			}else {
-			x+=al.get()*1000000000.0/speed;
-			}
+			x+=al*1000000000L/speed;
+			}*/
 
 			/*System.out.println(klalbRemoteSocket);
 			System.out.println(klalbRemoteSocket.getSendQueue().size());
@@ -36,14 +32,14 @@ public class LineDecitionComparator implements Comparator<KLALBRemoteSocket> {
 			predictedlatencys.put(klalbRemoteSocket, x);
 		}
 	}
-	public Map<KLALBRemoteSocket, Double> getPredictedlatencys() {
+	public Map<KLALBRemoteLine, Long> getPredictedlatencys() {
 		return predictedlatencys;
 	}
 
 	@Override
-	public int compare(KLALBRemoteSocket o1, KLALBRemoteSocket o2) {	
-		double t1=predictedlatencys.get(o1);
-		double t2=predictedlatencys.get(o2);
+	public int compare(KLALBRemoteLine o1, KLALBRemoteLine o2) {	
+		long t1=predictedlatencys.get(o1);
+		long t2=predictedlatencys.get(o2);
 		if(t1>t2) {
 			return 1;
 		}else if(t1<t2){

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

@@ -2,12 +2,21 @@ package org.kne.cloud.network.klalb;
 
 import java.util.Timer;
 import java.util.TimerTask;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Consumer;
 
 import static org.kne.cloud.network.klalb.KLALBUtils.*;
 public class Monitor {
 
+	private String name;
+	
+	public String getName() {
+		return name;
+	}
+	public void setName(String name) {
+		this.name = name;
+	}
 	private static Timer t=new Timer("带宽测量线程",true);
 	
 	
@@ -26,8 +35,13 @@ public class Monitor {
 	protected void finalize() throws Throwable {
 		ptt.cancel();
 	}
+	private static AtomicInteger ai=new AtomicInteger();
 	public Monitor() {
-
+		this.name="Line"+ai.getAndIncrement();
+		t.scheduleAtFixedRate(ptt, 1000, 1000);
+	}
+	public Monitor(String name) {
+		this.name=name;
 		t.scheduleAtFixedRate(ptt, 1000, 1000);
 	}
 	private AtomicLong inTraffic=new AtomicLong();
@@ -40,16 +54,31 @@ public class Monitor {
 	private volatile long inSpeed;
 	private volatile long outSpeed;
 	
-	private volatile long inSpeedMax;
-	private volatile long outSpeedMax;
 
-	private volatile long latency ;
+	private volatile long inSpeedAvg;
+	private volatile long outSpeedAvg;
+	
+	
+	private volatile long inSpeedMax=1024*1024*1024;
+	private volatile long outSpeedMax=1024*1024*1024;
+	
+	public void setInSpeedMax(long inSpeedMax) {
+		this.inSpeedMax = inSpeedMax;
+	}
+	public void setOutSpeedMax(long outSpeedMax) {
+		this.outSpeedMax = outSpeedMax;
+	}
+	private volatile long latencyAvg =Long.MIN_VALUE;
+	private volatile long latencyMin =Long.MIN_VALUE;
+	private volatile long latencyCurr =Long.MIN_VALUE;
 	private volatile long jitter ;
 	
 	private volatile int state=0;
 	public static final int OFFLINE=0;
 	public static final int CONNECTING=1;
 	public static final int ONLINE=2;
+	
+	
 	private volatile long coolingTime;
 	{
 		resetCoolingTime();
@@ -65,19 +94,49 @@ public class Monitor {
 		return inSpeedMax;
 	}
 	public void updateLatency(long newlatency) {
-		long vj=Math.abs( latency-newlatency);
+			latencyCurr=newlatency;
+		if(latencyAvg==Long.MIN_VALUE) {
+			latencyAvg=newlatency;
+		}else {
+		long vj=Math.abs( latencyAvg-newlatency);
 			jitter=(9*jitter+vj)/10;
-			
-		/*	if(newlatency<latency) {
+			/*if(newlatency<latency) {
 			latency=newlatency;
 		}else {*/
-		this.latency=(latency+ newlatency)>>1;
+		this.latencyAvg=(latencyAvg*9+ newlatency)/10;
 		//}
 		}
+		
+		if(latencyMin==Long.MIN_VALUE) {
+			latencyMin=newlatency;
+		}else {
+			/*if(newlatency<latency) {
+			latency=newlatency;
+		}else {*/
+			if(newlatency<=latencyMin) {
+				latencyMin=newlatency;
+			}else {
+		this.latencyMin=(latencyMin*9+ newlatency)/10;
+			}
+		//}
+		}
+	}
 	public long getOutSpeedMax() {
 		return outSpeedMax;
 	}
 	
+	public long getInSpeedAvg() {
+		return inSpeedAvg;
+	}
+	public long getOutSpeedAvg() {
+		return outSpeedAvg;
+	}
+	public long getLatencyMin() {
+		return latencyMin;
+	}
+	public long getLatencyCurr() {
+		return latencyCurr;
+	}
 	public long getInTraffic() {
 		return inTraffic.get();
 	}
@@ -119,7 +178,7 @@ public class Monitor {
 		outSpeed= o*1000000000/d;
 		updatetime=System.nanoTime();
 		//System.out.println(d+" "+o+" "+i);
-		if(inSpeed>inSpeedMax) {
+		/*if(inSpeed>inSpeedMax) {
 			inSpeedMax=inSpeed;
 		}else {
 			inSpeedMax=(inSpeedMax*9999+inSpeed)/10000;
@@ -128,11 +187,22 @@ public class Monitor {
 			outSpeedMax=outSpeed;
 		}else {
 			outSpeedMax=(outSpeedMax*9999+outSpeed)/10000;
-		}
+		}*/
+
+		this.inSpeedAvg=(inSpeedAvg*9+ inSpeed)/10;
+		this.outSpeedAvg=(outSpeedAvg*9+ outSpeed)/10;
 		
 		if(changeListener!=null) {
 			changeListener.accept(this);
 		}
+		
+	}
+	public void updateOutSpeedMax() {
+		if(outSpeed>outSpeedMax) {
+			outSpeedMax=outSpeed;
+		}else {
+			outSpeedMax=(outSpeedMax*9999+outSpeed)/10000;
+		}
 	}
 	private Consumer<Monitor>changeListener;
 	
@@ -142,22 +212,26 @@ public class Monitor {
 	public void setChangeListener(Consumer<Monitor> changeListener) {
 		this.changeListener = changeListener;
 	}
-	public long getLatency() {
-		return latency;
+	public long getLatencyAvg() {
+		return latencyAvg;
 	}
 	public String toString() {
 		StringBuilder sb=new StringBuilder();
+		sb.append(name);
+		sb.append('\t');
 		sb.append(getDsc());
 		sb.append('\t');
-		sb.append(bytesUnit(outTraffic.get())).append("\u2191\t").append(bytesUnit(inTraffic.get())).append("\u2193\t").append(bytesUnit(outSpeed)).append("/s\u2191\t").append(bytesUnit(inSpeed)).append("/s\u2193\t").append(latency/1000000).append("ms");
+		sb.append(bytesUnit(outTraffic.get())).append("\u2191\t").append(bytesUnit(inTraffic.get())).append("\u2193\t").append(bytesUnit(outSpeed)).append("/s\u2191\t").append(bytesUnit(inSpeed)).append("/s\u2193\t").append(latencyAvg/1000000).append("ms");
 		return sb.toString();
 	}
 	
 	public String toString2() {
 		StringBuilder sb=new StringBuilder();
+		sb.append(name);
+		sb.append('\n');
 		sb.append(getDsc());
 		sb.append('\t');
-		sb.append(bytesUnit(outTraffic.get())).append("\t").append(bytesUnit(inTraffic.get())).append("\t").append(bytesUnit(outSpeed)).append("/s\t").append(bytesUnit(inSpeed)).append("/s\t").append(latency/1000000).append("ms").append("\t").append(jitter/1000000).append("ms\t");
+		sb.append(bytesUnit(outTraffic.get())).append("\t").append(bytesUnit(inTraffic.get())).append("\t").append(bytesUnit(outSpeed)).append("/s\t").append(bytesUnit(inSpeed)).append("/s\t").append(latencyAvg/1000000).append("ms").append("\t").append(jitter/1000000).append("ms\t");
 		
 		if(state==OFFLINE) {
 		sb.append((coolingTime-(System.currentTimeMillis()-mls))/1000L);

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

@@ -8,7 +8,7 @@ import java.util.Timer;
 import java.util.TimerTask;
 import java.util.concurrent.atomic.AtomicLong;
 
-import org.kne.cloud.network.mport.FilterSocket;
+import org.kne.cloud.network.FilterSocket;
 
 public class MonitoredSocket extends FilterSocket {
 	public Monitor getMonitor() {

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

@@ -5,39 +5,53 @@ import java.io.DataOutput;
 import java.io.IOException;
 
 public class PONGPacket extends KLALBPacket {
-	public long getTime() {
-		return time;
+	
+	public long getTimepingsnd() {
+		return timepingsnd;
 	}
-	private long time;
+
+	public long getTimepingrcv() {
+		return timepingrcv;
+	}
+
+	public long getTimepongsnd() {
+		return timepongsnd;
+	}
+	private long timepingsnd,timepingrcv,timepongsnd;
 
 	@Override
 	protected void writeToStream(DataOutput dto) throws IOException {
 		super.writeToStream(dto);
-		dto.writeLong(time);
+		dto.writeLong(timepingsnd);
+		dto.writeLong(timepingrcv);
+		dto.writeLong(timepongsnd);
 	}
 
 	@Override
 	protected void readFromStream(DataInput din) throws IOException {
 		super.readFromStream(din);
-		time=din.readLong();
+		timepingsnd=din.readLong();
+		timepingrcv=din.readLong();
+		timepongsnd=din.readLong();
 	}
 
 	@Override
 	public long getLength() {
-		return super.getLength()+8;
+		return super.getLength()+24;
 	}
 	public PONGPacket() {
 		super(PONG);
 	}
-	public PONGPacket(long time) {
+	
+	public PONGPacket( long timepingsnd, long timepingrcv, long timepongsnd) {
 		super(PONG);
-		this.time=time;
+		this.timepingsnd = timepingsnd;
+		this.timepingrcv = timepingrcv;
+		this.timepongsnd = timepongsnd;
 	}
+
 	@Override
 	public String toString() {
 		return "PONG";
 	}
-	public void redeltaTime(long delta) {
-		time+=delta;
-	}
 }

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

@@ -4,7 +4,7 @@ import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
 
-public class RSTPacket extends KLALBPacket {
+public class RSTPacket extends KLALBPacket implements PortPacket{
 	private int sport,dport;
 	public RSTPacket(int sport,int dport) {
 		super(RST);

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

@@ -58,8 +58,7 @@ public class SendTask {
 
 	
 	public void check(int number) throws IOException {
-		long limit= (1<<(2*Math.min(4,count.get()-1)))*(number<3?200:1000*number)*1000000L;
-		//long limit=(1<<Math.min(4,count.get()-1)*1000L*number+1000L)*1000000L;
+		long limit= count.get()*(1000*number+500)*1000000L;
 		if(checkTime(limit))
 			run();
 	}
@@ -72,10 +71,10 @@ public class SendTask {
 		try {
 			if(count.get()==0) {
 			controllker.sendPacketToAddress((Inet6Address) address,
-					packet, priority+count.get(),2);
+					packet, priority,1);
 			}else {
 				controllker.sendPacketToAddress((Inet6Address) address,
-						packet, priority+count.get(),2);
+						packet, priority-1,1);
 			}
 		} catch (NoRouteToHostException e) {
 		}

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

@@ -2,6 +2,8 @@ package org.kne.cloud.network.klalb;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.Inet6Address;
+import java.net.InetSocketAddress;
 import java.net.ServerSocket;
 import java.net.UnknownHostException;
 import java.util.Iterator;
@@ -9,67 +11,37 @@ import java.util.List;
 import java.util.Properties;
 import java.util.Scanner;
 
-import org.kne.cloud.network.klalb.LineManager.LineEntry;
-import org.kne.cloud.network.mport.MultipurposeSocketAddress;
-import org.kne.cloud.network.mport.SocketBridge;
-import org.kne.cloud.network.mport.TCPListener;
+import org.kne.cloud.network.MultipurposeSocketAddress;
+import org.kne.cloud.network.SocketBridge;
+import org.kne.cloud.network.SocketListener;
+import org.kne.cloud.network.SocketToSocketProxy;
 import org.kne.util.AutoProperties;
 
 public class SimpleKLALBClient {
 public static void main(String[] args) throws UnknownHostException, IOException {
+
+	System.out.println(CONST.klalb+" V"+CONST.klalbver);
 	Scanner scn=new Scanner(System.in);
 	Properties def=new Properties();
 	def.setProperty("server", "");
 	def.setProperty("local", "");
 	AutoProperties ap=new AutoProperties(new File("klalbclient.ini"),def);
 	KLALBController kc=new KLALBController();
+	kc.registerToProxyTypeAs("KLALB");
 	MultipurposeSocketAddress msa=new MultipurposeSocketAddress(ap.getProperty("server"));
-	KLALBRemoteSocket kr=new KLALBRemoteSocket(msa.connectSocket());
-	kc.addRemoteSocket(kr);
-	kr.close();
+	Inet6Address vad= kc.getRemoteVaddrBySocketAddress(msa);
+	MultipurposeSocketAddress vmsa=new MultipurposeSocketAddress("KLALB_Stream",new InetSocketAddress(vad, 23333));
 	System.out.println("连接成功："+ap.getProperty("server"));
-	kc.getLineManager().addHostPort(msa);
-	TCPListener tl=new TCPListener(new MultipurposeSocketAddress(ap.getProperty("local")));
-	tl.setCon((s)->{
-		KLALBVirtualSocket kvs=null;
-		try {
-		s.setTcpNoDelay(true);
-			kvs=new KLALBVirtualSocket(kc, kr.getRemoteVaddr(), 23333);
-			kvs.setTcpNoDelay(true);
-			SocketBridge dsb=new SocketBridge(s, kvs);
-			dsb.getSab().setDelay(1);
-			dsb.run();
-			/*CompressedSocketBridge csb=new CompressedSocketBridge(s, kvs);
-			csb.getSab().setDelay(1);
-			csb.run();*/
-		} catch (IOException e) {
-			e.printStackTrace();
-		}finally {
-			
-			try {
-				s.close();
-			} catch (IOException e) {
-				e.printStackTrace();
-			}
-			if(kvs!=null)
-				try {
-					kvs.close();
-				} catch (IOException e) {
-					e.printStackTrace();
-				}
-		}
-	});
-	tl.open();
+	new SocketToSocketProxy(new MultipurposeSocketAddress(ap.getProperty("local")), vmsa);
 	System.out.println("提示：输入state并回车可以查看当前线路状态");
 	while(true) {
 		String s=scn.nextLine();
 		switch(s) {
 		case "state":
 			System.out.println("状态\t上传流量\t下载流量\t上传速度\t下载速度\t延迟\t抖动\t下一次重试");
-			LineManager le=kc.getLineManager();
-			for (Iterator<java.util.Map.Entry<MultipurposeSocketAddress, LineEntry>> iterator = le.entrySet().iterator(); iterator.hasNext();) {
-				java.util.Map.Entry<MultipurposeSocketAddress, LineEntry> hostPort =  iterator.next();
-				System.out.println(hostPort.getValue() .toString2());
+			for (Iterator<KLALBRemoteLine> iterator = kc.getLines().iterator(); iterator.hasNext();) {
+				KLALBRemoteLine hostPort =  iterator.next();
+				System.out.println(hostPort .toString2());
 				//System.out.println();
 			}
 			break;

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

@@ -4,34 +4,33 @@ import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileReader;
 import java.io.IOException;
-import java.net.InetAddress;
-import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.UnknownHostException;
 import java.util.Iterator;
-import java.util.List;
 import java.util.Scanner;
 
-import org.kne.cloud.network.klalb.LineManager.LineEntry;
-import org.kne.cloud.network.mport.MultipurposeSocketAddress;
-import org.kne.cloud.network.mport.Protocol;
-import org.kne.cloud.network.mport.ProtocolDetectorServerSocketFactory;
-import org.kne.cloud.network.mport.ProtocolDetectorSocket;
-import org.kne.cloud.network.mport.ProxyProfileAnalyser;
-import org.kne.cloud.network.mport.ProxyProfileExecutor;
-import org.kne.cloud.network.mport.SocketBridge;
-import org.kne.cloud.network.mport.TCPListener;
+import org.kne.cloud.network.DatagramSocketListener;
+import org.kne.cloud.network.MultipurposeSocketAddress;
+import org.kne.cloud.network.ProtocolDetectorServerSocketFactory;
+import org.kne.cloud.network.ProtocolDetectorSocket;
+import org.kne.cloud.network.SocketBridge;
+import org.kne.cloud.network.SocketListener;
+import org.kne.cloud.network.SocketToSocketProxy;
+import org.kne.cloud.network.SocketType;
 
 public class SimpleKLALBServer {
-	public static TCPListener tcpl,tcpl2;
+	public static SocketListener tcpl;
+	public static DatagramSocketListener udpl;
 	public static ServerPropties sp;
 	public static KLALBController kc;
 	static {
-		MultipurposeSocketAddress.getServerSocketFactoryRegister().put("DETTCP", new ProtocolDetectorServerSocketFactory());
+		MultipurposeSocketAddress.getSocketTypeRegister().put("DETTCP",new SocketType(null, new ProtocolDetectorServerSocketFactory()));
 	}
 	public static void main(String[] args) throws IOException {
+		//Debuger dbg=new Debuger();
+		//dbg.start();
 		
-		System.out.println("KLALB负载均衡V2.0");
+		System.out.println(CONST.klalb+" V"+CONST.klalbver);
 		sp=new ServerPropties();
 		
 		System.out.println("虚拟地址："+sp.getVirtualIP().getHostAddress());
@@ -58,10 +57,11 @@ public class SimpleKLALBServer {
 			System.out.println("reload:重新加载线路配置");
 		break;
 		case "state":
-			LineManager le=kc.getLineManager();
-			for (Iterator<java.util.Map.Entry<MultipurposeSocketAddress, LineEntry>> iterator = le.entrySet().iterator(); iterator.hasNext();) {
-				java.util.Map.Entry<MultipurposeSocketAddress, LineEntry> hostPort =  iterator.next();
-				System.out.println(hostPort.getValue() .toString());
+			System.out.println("状态\t上传流量\t下载流量\t上传速度\t下载速度\t延迟\t抖动\t下一次重试");
+			for (Iterator<KLALBRemoteLine> iterator = kc.getLines().iterator(); iterator.hasNext();) {
+				KLALBRemoteLine hostPort =  iterator.next();
+				System.out.println(hostPort .toString2());
+				//System.out.println();
 			}
 			break;
 		default:
@@ -72,13 +72,21 @@ public class SimpleKLALBServer {
 	private static void openPort(String bip) throws IOException {
 		if(tcpl!=null)
 			tcpl.close();
+		if(udpl!=null) {
+			udpl.close();
+		}
 		MultipurposeSocketAddress mpsa=new MultipurposeSocketAddress(bip,"DETTCP");
-		tcpl=new TCPListener(mpsa);
+		tcpl=new SocketListener(mpsa);
 		tcpl.setCon((soc)->{
 			ProtocolDetectorSocket pds=(ProtocolDetectorSocket) soc;
 		if(!pds.getProtocolStack().isEmpty()&&pds.getProtocolStack().pop().getName().equals("KLALB")) {
-			KLALBRemoteSocket krs=new KLALBRemoteSocket(pds);
-			kc.addRemoteSocket(krs);
+			KLALBRemoteLine krs=null;
+			try {
+				krs = new KLALBRemoteLine(new StreamKLALBPacketLink(pds));
+			kc.addRemoteLine(krs);
+			} catch (IOException e) {
+				e.printStackTrace();
+			}
 			}else {
 				MultipurposeSocketAddress mpsa2=new MultipurposeSocketAddress(sp.getLocal());
 				Socket s=null;
@@ -108,49 +116,21 @@ public class SimpleKLALBServer {
 				}
 			}
 		});
-		tcpl.open();
+		udpl=new DatagramSocketListener(new MultipurposeSocketAddress(bip, "UDP"));
+		udpl.setCon((r)->{
+			KLALBRemoteLine krl;
+			try {
+				krl=new KLALBRemoteLine(new DatagramKLALBPacketLink(r));
+				kc.addRemoteLine(krl);
+			} catch (IOException e) {
+				// TODO 自动生成的 catch 块
+				e.printStackTrace();
+			}
+		});
 	}
 	private static void openLocalPort(String bip) throws IOException {
-		if(tcpl2!=null)
-			tcpl2.close();
 		MultipurposeSocketAddress mpsa=new MultipurposeSocketAddress(bip);
-		tcpl2=new TCPListener(new MultipurposeSocketAddress("KLALB", "::0", 23333));
-		tcpl2.setCon((soc)->{
-			Socket s=null;
-			try {
-				s=mpsa.connectSocket();
-				s.setTcpNoDelay(true);
-				soc.setTcpNoDelay(true);
-				SocketBridge dsb=new SocketBridge(s, soc);
-				dsb.getSab().setDelay(1);
-				dsb.run();
-				/*CompressedSocketBridge csb=new CompressedSocketBridge(s, soc);
-				csb.getSab().setDelay(1);
-				csb.run();*/
-			} catch (UnknownHostException e) {
-				// TODO 自动生成的 catch 块
-				e.printStackTrace();
-			} catch (IOException e) {
-				// TODO 自动生成的 catch 块
-				e.printStackTrace();
-			}finally {
-				if(s!=null)
-					try {
-						s.close();
-					} catch (IOException e1) {
-						// TODO 自动生成的 catch 块
-						e1.printStackTrace();
-					}
-				try {
-					soc.close();
-				} catch (IOException e) {
-					// TODO 自动生成的 catch 块
-					e.printStackTrace();
-				}
-			}
-			
-		});
-		tcpl2.open();
+		new SocketToSocketProxy(new MultipurposeSocketAddress("KLALB_Stream", "::0", 23333),mpsa);
 	}
 	  private static String fileRead(String filePath){
 	        //1.定义一个BufferedReader对象，将文件内容读取到缓存

src/org/kne/cloud/network/mport/SocketToSocketProxyProfileEntry.java

@@ -1,16 +0,0 @@
-package org.kne.cloud.network.mport;
-
-public class SocketToSocketProxyProfileEntry extends ProxyProfileEntry {
-	public SocketToSocketProxyProfileEntry(String l, String r) {
-		src=new MultipurposeSocketAddress(l);
-		des=new MultipurposeSocketAddress(r);
-	}
-	private MultipurposeSocketAddress src;
-	private MultipurposeSocketAddress des;
-	public MultipurposeSocketAddress getSrc() {
-		return src;
-	}
-	public MultipurposeSocketAddress getDes() {
-		return des;
-	}
-}

src/org/kne/cloud/network/nathole/NatholeTestS.java

@@ -9,12 +9,12 @@ import java.nio.channels.SocketChannel;
 
 public class NatholeTestS {
 	public static void main(String[] args) throws IOException {
-		for (int i1 = 1024; i1 < 65536; i1++) {
+		for (int i1 = 1024; i1 < 32768; i1++) {
 			try {
 			SocketChannel sc=	SocketChannel.open();
 		sc.configureBlocking(false)	;
 		sc.bind(new InetSocketAddress("0.0.0.0", i1));
-		sc.connect(new InetSocketAddress("10.235.16.1",10300));
+		sc.connect(new InetSocketAddress("183.199.49.116",21000));
 		sc.close();
 		int i2=i1;
 		new Thread(()->{

src/org/kne/debug/TimeDebugger.java

@@ -3,15 +3,16 @@ package org.kne.debug;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Hashtable;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 
 public class TimeDebugger {
-	Map<String,Long> l=new Hashtable<String,Long>();
+	Map<String,Long> l=new LinkedHashMap<String,Long>();
 	long tmp=-1;
 	public void putTime(String name){
 		
-		long v=System.currentTimeMillis();
+		long v=System.nanoTime();
 		if(tmp==-1){
 			l.put(name,  0l);
 		}else{
@@ -20,6 +21,6 @@ public class TimeDebugger {
 		tmp=v;
 	}
 	public void print(){
-		System.err.println(l+"(ms)");
+		System.err.println(l+"(ns)");
 	}
 }