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

package org.kne.cloud.network.klalb;

import java.io.IOException;
import java.net.Inet6Address;
import java.net.NoRouteToHostException;
import java.net.Socket;
import java.net.SocketException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;

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

public class KLALBRemoteSocket extends MonitoredSocket {
	private Inet6Address remoteVaddr;

	private KLALBController controller;
	
	public KLALBController getController() {
		return controller;
	}
	protected void setController(KLALBController controller) {
		this.controller = controller;
	}
	public Inet6Address getRemoteVaddr() {
		return remoteVaddr;
	}
	public String toString() {
		return super.toString()+ getMonitor().toString();
	}
public KLALBRemoteSocket(Socket socket) {
	this(socket,new Monitor());
}
	public KLALBRemoteSocket(Socket socket,Monitor m) {
		super(socket,m);
		try {
			//socket.setSendBufferSize(32768);
			//socket.setReceiveBufferSize(65535);
			//System.out.println(socket.getSendBufferSize()+" "+socket.getReceiveBufferSize());
			socket.setTcpNoDelay(true);
			socket.setSoTimeout(10000);
		} catch (SocketException e1) {
			e1.printStackTrace();
		}
		ThreadTool.makeVDaemonThreadIfSupport("远程接收线程", () -> {
			try {

				while (!isClosed()) {
					KLALBPacket kpp = getInputStream().readPacket();
						if(kpp instanceof PINGPacket) {
						sendPacket(new PONGPacket(((PINGPacket) kpp).getTime()), 65540);
					}else if(kpp instanceof PONGPacket) {
						PONGPacket png=(PONGPacket) kpp;
						getMonitor().updateLatency (System.nanoTime()- png.getTime());
						try {
							socket.setSoTimeout(600);
						} catch (SocketException e1) {
							// TODO 自动生成的 catch 块
							e1.printStackTrace();
						}
					}else {
					
					System.out.println("RX:" + kpp);
						if(kpp instanceof VADDRPacket) {
							remoteVaddr=((VADDRPacket) kpp).getVaddr();
						}
						while(rec==null) {
							Thread.sleep(1);
						}
					rec.accept(kpp);
					}
				}
			} catch (IOException e) {
				//e.printStackTrace();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} finally {
				try {
					close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}).start();
		ThreadTool.makeVDaemonThreadIfSupport("远程发送线程", () -> {
			try {

				while (!isClosed()) {
					if(checkPingTime()) {
						getOutputStream().writePacket(new PINGPacket(System.nanoTime()));
					}else {
					PQItem pqi=sendQueue.poll();
					if(pqi!=null) {
					KLALBPacket kpp=pqi.getPacket();
					if(!(kpp instanceof PONGPacket))
					System.out.println("TX:" + kpp);
					if(kpp instanceof PONGPacket) {
						PONGPacket pp=(PONGPacket) kpp;
						pp.redeltaTime(System.nanoTime()-pqi.getAddtime());
					}
					getOutputStream().writePacket(kpp);
					//System.out.println(sendQueue.size());
					}else {
						synchronized (sendQueue) {
						sendQueue.wait(10);
						}
							//Thread.sleep(1);
					}
					}
				}
			} catch (IOException e) {
			//	e.printStackTrace();
			} catch (InterruptedException e) {
			} finally {
				try {
					close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}).start();
	}
	

	private volatile long time = System.nanoTime();
	private volatile long pingInterval=500000000L;

	public long getPingInterval() {
		return pingInterval;
	}

	public void setPingInterval(long ns) {
		this.pingInterval = ns;
	}

	private boolean checkPingTime() {
		long cu = System.nanoTime();
		if (cu - time > pingInterval) {
			time = cu;
			return true;
		} else {
			return false;
		}
	}
	




	private KLALBInputStream kis;
	private KLALBOutputStream kos;

	@Override
	public KLALBInputStream getInputStream() throws IOException {
		if (kis == null)
			kis = new KLALBInputStream(super.getInputStream());
		return kis;
	}

	@Override
	public KLALBOutputStream getOutputStream() throws IOException {
		if (kos == null)
			kos = new KLALBOutputStream(super.getOutputStream());
		return kos;
	}

	
	private volatile Consumer<KLALBPacket> rec;
	private Consumer<KLALBRemoteSocket> clo;
	private PriorityBlockingQueue<PQItem> sendQueue=new PriorityBlockingQueue<PQItem>();
	
	protected PriorityBlockingQueue<PQItem> getSendQueue() {
		return sendQueue;
	}

	public void sendPacket(KLALBPacket kp) {
		sendPacket(kp, 5);
	}
	
	public void sendPacket(KLALBPacket kp,int priority) {
		sendQueue.add(new PQItem(kp, priority));
		synchronized (sendQueue) {
			sendQueue.notifyAll();
		}
	}

	public void setPacketReceiver(Consumer<KLALBPacket> rec) {
		this.rec = rec;
	}
	public void setCloseListener(Consumer<KLALBRemoteSocket> clo) {
		this.clo = clo;
	}
	@Override
	public synchronized void close() throws IOException {
		if(!isClosed()&&clo!=null)
			clo.accept(this);
		cdl.countDown();
		super.close();
		sendQueue.forEach((x)->{
			try {
				//System.out.println("断线重发");
				controller.sendPacketToAddress(remoteVaddr, x.getPacket(), x.getPriority()+1);
			} catch (NoRouteToHostException e) {
			}
		});
	}

	@Override
	public boolean isClosed() {
		return cdl.getCount()<=0;
	}

	private AtomicLong al=new AtomicLong(0);
	protected class PQItem implements Comparable<PQItem>{
		private KLALBPacket packet;
		private long index=al.getAndIncrement();
		private int priority=5;
		private long addtime=System.nanoTime();
		public long getAddtime() {
			return addtime;
		}
		public PQItem(KLALBPacket value, int priority) {
			super();
			this.packet = value;
			this.priority = priority;
		}
		public KLALBPacket getPacket() {
			return packet;
		}
		public long getIndex() {
			return index;
		}
		public int getPriority() {
			return priority;
		}
		@Override
		public int compareTo(PQItem o) {
			if(priority<o.priority) {
				return 1;
			}else if(priority>o.priority) {
				return -1;
			}else {
				if(index>o.index) {
					return 1;
				}else if(index<o.index){
					return -1;
				}else {
					return 0;
				}
			}
		}
	}
	private CountDownLatch cdl=new CountDownLatch(1);
	public void waitforlose() {
		try {
			cdl.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	public void remoeFromSendQueue(KLALBPacket klalbPacket) {
		sendQueue.removeIf((x)->{
			return x.getPacket().equals(klalbPacket);
		});
	}
}