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

package org.kne.cloud.network.klalb;

import java.io.IOException;
import java.io.StreamCorruptedException;
import java.net.InetSocketAddress;
import java.net.SocketException;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.Channels;
import java.nio.channels.GatheringByteChannel;
import java.nio.channels.ScatteringByteChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicLong;

import org.kne.cloud.network.BufferedChannel;
import org.kne.cloud.network.MultipurposeSocketAddress;
import org.kne.cloud.network.NetworkPacket;
import org.kne.cloud.network.ThreadTool;
import org.kne.io.KNEChannels;

public class StreamChannelKLALBPacketLink extends AbstractKLALBPacketLink implements KLALBPacketLink {
	private static final boolean debug = false;
	private volatile long timeoutTimer;
	private volatile boolean timerenabled=false;
	private boolean enableBuffer=true;
	private Runnable timeouter=new Runnable() {
		public void run() {
			timeoutTimer=System.nanoTime();
			while(connectSocket.isOpen()) {
				if(timerenabled&&(System.nanoTime()-timeoutTimer>sotimeout*1000000L)) {
					if(connectSocket!=null)
						try {
							connectSocket.close();
						} catch (IOException e) {
						}
				}
				try {
					Thread.sleep(5);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		}
	};
	@Override
	public String toString() {
		try {
			return new MultipurposeSocketAddress("TCP",(InetSocketAddress)connectSocket.getLocalAddress())+"←"+new MultipurposeSocketAddress("TCP",(InetSocketAddress)connectSocket.getRemoteAddress());
		} catch (Exception e) {
			e.printStackTrace();
			return "?←?";
		}
	}

	private SocketChannel connectSocket;
	private ScatteringByteChannel readableChannel;
	private GatheringByteChannel writableChannel;
	private int sotimeout=getDefaultSoTimeout();
	public StreamChannelKLALBPacketLink(SocketChannel connectSocket) throws IOException {
		this(connectSocket,true);
	}
	public StreamChannelKLALBPacketLink(SocketChannel connectSocket,boolean enableBuffer) throws IOException {
		try {
		this.connectSocket=connectSocket;
		if(enableBuffer) {
			BufferedChannel buf=new BufferedChannel(connectSocket,connectSocket,256*1024);
			this.readableChannel =buf;
			this.writableChannel =buf;
		}else {
			this.readableChannel=connectSocket;
			this.writableChannel=connectSocket;
		}
		this.enableBuffer=enableBuffer;
		connectSocket.setOption(StandardSocketOptions.TCP_NODELAY,true);
		ThreadTool.makeVDaemonThread("连接超时计时线程", timeouter);
		new KLALBOutputStream(Channels.newOutputStream(writableChannel));
		if(writableChannel instanceof BufferedChannel)
			((BufferedChannel) writableChannel).flush();
		new KLALBInputStream(Channels.newInputStream(readableChannel));
		}catch(Throwable e) {
			connectSocket.close();
			throw e;
		}
	}
	
	
	

	@Override
	public void close() throws IOException {
		//System.out.println("连接被关闭");
		//new Exception().printStackTrace();
		connectSocket.close();
	}

	@Override
	public boolean isClosed() {
		return !connectSocket.isOpen();
	}

	@Override
	public void setSoTimeout(int val) throws SocketException {
		sotimeout=val;
	}

	@Override
	public int getSoTimeout() throws SocketException {
		return	sotimeout;
	}

	@Override
	public void flush() throws IOException {
		checkflush();
	}

	@Override
	public boolean isStream() {
		return true;
	}

	

	@Override
	public void writePacket(ByteBuffer kp) throws IOException {
		ByteBuffer szeWrite=NetworkPacket.bufferAllocator.allocate(4);
		szeWrite.limit(4);
		//szeWrite.clear();
		szeWrite.putInt(kp.limit());
		szeWrite.flip();
		KLALBVirtualSocketChannel obj = null;
		writableChannel.write(new ByteBuffer[] {szeWrite,kp});
		checkflush();
		
		
	}
    
	


	@Override
	public ByteBuffer readPacket() throws IOException {

		ByteBuffer szeRead=NetworkPacket.bufferAllocator.allocate(4);
		timeoutTimer=System.nanoTime();
		timerenabled=true;
		ByteBuffer kp;
		szeRead.clear();
		szeRead.limit(4);
		//szeRead.clear();
		try {
			KNEChannels.readFully(readableChannel,szeRead);
			szeRead.flip();
			int size=szeRead.getInt(0);
			kp=NetworkPacket.bufferAllocator.allocate(size);
			kp.limit(size);
			KNEChannels.readFully(readableChannel, kp);
			kp.flip();
		return kp;
		}finally {
			timerenabled=false;
		}
	}
	
	
	
	
	@Override
	public void writeKLALBPacket(KLALBPacket kp) throws IOException {
	
		/*ByteBuffer szeWrite=NetworkPacket.bufferAllocator.allocate(4);
		szeWrite.limit(4);
		//szeWrite.clear();
		int length=(int) kp.getTotalLength();
		szeWrite.putInt(0,length);
		szeWrite.flip();
		
		
		
		writableChannel.write(szeWrite);*/

		if(debug) {

			int length=(int) kp.getTotalLength();
			AtomicLong al=new AtomicLong(0);
			KLALBPacket.writeKLALBPacketToChannel(new MonitoredChannel(writableChannel,null,new AtomicLong[] { al}), kp);
			if(length!=al.get()) {
				throw new StreamCorruptedException(kp+" packet length error:"+al.get()+"!="+length);
			}
		}else {
			KLALBPacket.writeKLALBPacketToChannel(writableChannel, kp);
		}
		
		
	}

	private void checkflush() throws IOException {
		if(writableChannel instanceof BufferedChannel) {
			((BufferedChannel) writableChannel).flush();
		}else if(writableChannel instanceof KLALBVirtualSocketChannel) {
			((KLALBVirtualSocketChannel) writableChannel).flush();
			
		}
	}

	@Override
	public KLALBPacket readKLALBPacket() throws IOException {
	
		/*ByteBuffer szeWrite=NetworkPacket.bufferAllocator.allocate(4);
		szeWrite.limit(4);
		KNEChannels.readFully(readableChannel, szeWrite);
		szeWrite.flip();
		int readSize=szeWrite.getInt();*/
		KLALBPacket kp=KLALBPacket.readKLALBPacketFromChannel(readableChannel);
	if(debug) {
		int readSize=0;
		long kpl=kp.getTotalLength();
		if(kpl!=readSize) {
			throw new StreamCorruptedException(kp+" packet length error:"+kpl+"!="+readSize);
		}
	}
		return kp;
	}

	
	
}