forked from KNEMC/KLALB
KLALB完全大重写，代码规范了，可以当作网络库使用
This commit is contained in:
Administrator
@@ -0,0 +1,503 @@
|
||||
package org.kne.cloud.network.klalb;
|
||||
|
||||
import java.io.ByteArrayOutputStream;
|
||||
import java.io.FileDescriptor;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStream;
|
||||
import java.io.OutputStream;
|
||||
import java.net.BindException;
|
||||
import java.net.ConnectException;
|
||||
import java.net.Inet4Address;
|
||||
import java.net.Inet6Address;
|
||||
import java.net.InetAddress;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.NoRouteToHostException;
|
||||
import java.net.SocketAddress;
|
||||
import java.net.SocketException;
|
||||
import java.net.SocketImpl;
|
||||
import java.net.SocketOptions;
|
||||
import java.net.SocketTimeoutException;
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
import java.util.concurrent.BlockingDeque;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.LinkedBlockingDeque;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
import java.util.function.BiConsumer;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.kne.io.Data;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
public class KLALBVirtualSocketImpl extends SocketImpl {
|
||||
private KLALBController controller;
|
||||
|
||||
protected int getInputchachesize() {
|
||||
return inputchachesize;
|
||||
}
|
||||
|
||||
protected void setInputchachesize(int inputchachesize) {
|
||||
this.inputchachesize = inputchachesize;
|
||||
}
|
||||
|
||||
protected int getOutputchachesize() {
|
||||
return outputchachesize;
|
||||
}
|
||||
|
||||
protected void setOutputchachesize(int outputchachesize) {
|
||||
this.outputchachesize = outputchachesize;
|
||||
}
|
||||
|
||||
private int inputchachesize = 65535 * 50;
|
||||
private int outputchachesize = 65535 * 50;
|
||||
private Inet6Address bindaddr;
|
||||
|
||||
private BlockingQueue<InetSocketAddress> backlogQueue;
|
||||
private BlockingDeque<DATATPacket> sendDeque = new LinkedBlockingDeque<>();
|
||||
private List<SendTask> sendlist=new Vector<>();
|
||||
private boolean succeed, refused;
|
||||
|
||||
protected boolean isListening() {
|
||||
return backlogQueue != null;
|
||||
}
|
||||
|
||||
private List<DATATPacket> inputchache = new ArrayList<>();
|
||||
private long inputcount = 0;
|
||||
private boolean avaliable = true;
|
||||
private BiConsumer<Inet6Address, KLALBPacket> packReceiver = new BiConsumer<Inet6Address, KLALBPacket>() {
|
||||
|
||||
@Override
|
||||
public void accept(Inet6Address from, KLALBPacket u) {
|
||||
try {
|
||||
if (u instanceof SYNTPacket) {
|
||||
if (isListening()) {
|
||||
|
||||
if (backlogQueue.offer(new InetSocketAddress(from, ((SYNTPacket) u).getSport()))) {
|
||||
controller.sendPacketToAddress(from,
|
||||
new SACKTPacket(localport, ((SYNTPacket) u).getSport()), 65537);
|
||||
|
||||
} else {
|
||||
controller.sendPacketToAddress(from, new RSTPacket(localport, ((SYNTPacket) u).getSport()),
|
||||
65537);
|
||||
}
|
||||
} else {
|
||||
controller.sendPacketToAddress(from, new RSTPacket(localport, ((SYNTPacket) u).getSport()),
|
||||
65537);
|
||||
}
|
||||
} else if (u instanceof SACKTPacket) {
|
||||
if (connecting) {
|
||||
connecting = false;
|
||||
succeed = true;
|
||||
cdl.countDown();
|
||||
}
|
||||
} else if (u instanceof RSTPacket) {
|
||||
if (connecting) {
|
||||
connecting = false;
|
||||
refused = true;
|
||||
cdl.countDown();
|
||||
close();
|
||||
} else {
|
||||
close();
|
||||
}
|
||||
} else if (u instanceof DATATPacket) {
|
||||
DATATPacket dtp = (DATATPacket) u;
|
||||
synchronized (inputchache) {
|
||||
if (dtp.getNumber() >= inputcount) {
|
||||
inputchache.add(dtp);
|
||||
while (true) {
|
||||
DATATPacket kkb = null;
|
||||
for (int i = 0; i < inputchache.size(); i++) {
|
||||
DATATPacket klalbBlock = inputchache.get(i);
|
||||
if (klalbBlock.getNumber() == inputcount) {
|
||||
inputchache.remove(i);
|
||||
i--;
|
||||
kkb = klalbBlock;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (kkb == null)
|
||||
break;
|
||||
sendDeque.add(kkb);
|
||||
inputcount++;
|
||||
}
|
||||
}
|
||||
}
|
||||
controller.sendPacketToAddress(from, new ACKTPacket(dtp.getDport(), dtp.getSport(), dtp.getNumber(),
|
||||
countInputBytes() < inputchachesize), 32768);
|
||||
} else if (u instanceof ACKTPacket) {
|
||||
ACKTPacket ackt = (ACKTPacket) u;
|
||||
avaliable = ackt.isAvaliable();
|
||||
sendlist.removeIf((tsk)->{
|
||||
boolean b=tsk.getKp().getNumber()==ackt.getNumber();
|
||||
if(b)
|
||||
tsk.cancel();
|
||||
return b;
|
||||
});
|
||||
}
|
||||
} catch (IOException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
};
|
||||
private int countInputBytes() {
|
||||
AtomicInteger i = new AtomicInteger(0);
|
||||
sendDeque.forEach((c) -> {
|
||||
i.addAndGet(c.getData().length);
|
||||
});
|
||||
return i.get();
|
||||
}
|
||||
private int countOutputBytes() {
|
||||
AtomicInteger i = new AtomicInteger(0);
|
||||
sendlist.forEach((c) -> {
|
||||
i.addAndGet(c.getKp().getData().length);
|
||||
});
|
||||
return i.get();
|
||||
}
|
||||
public KLALBController getController() {
|
||||
return controller;
|
||||
}
|
||||
|
||||
public BiConsumer<Inet6Address, KLALBPacket> getPackReceiver() {
|
||||
return packReceiver;
|
||||
}
|
||||
|
||||
public KLALBVirtualSocketImpl(KLALBController kc) {
|
||||
super();
|
||||
this.controller = kc;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setOption(int optID, Object value) throws SocketException {
|
||||
// TODO 自动生成的方法存根
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object getOption(int optID) throws SocketException {
|
||||
// TODO 自动生成的方法存根
|
||||
return null;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void create(boolean stream) throws IOException {
|
||||
if (!stream) {
|
||||
throw new RuntimeException("please use KLALBVirtualDatagramSocket to process udp packet");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void connect(String host, int port) throws IOException {
|
||||
connect(InetAddress.getByName(host), port);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void connect(InetAddress address, int port) throws IOException {
|
||||
connect(new InetSocketAddress(address, port), 10000);
|
||||
}
|
||||
|
||||
private CountDownLatch cdl = new CountDownLatch(1);
|
||||
|
||||
@Override
|
||||
protected void connect(SocketAddress address, int timeout) throws IOException {
|
||||
if (!controller.checkIsBind(this)) {
|
||||
bind(Inet6Address.getByName("::0"), 0);
|
||||
}
|
||||
connecting = true;
|
||||
port = ((InetSocketAddress) address).getPort();
|
||||
this.address = ((InetSocketAddress) address).getAddress();
|
||||
controller.sendPacketToAddress((Inet6Address) this.address, new SYNTPacket(localport, port), 65537);
|
||||
try {
|
||||
if (timeout == 0) {
|
||||
cdl.await();
|
||||
} else {
|
||||
cdl.await(timeout, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
connecting = false;
|
||||
if (succeed) {
|
||||
|
||||
} else if (refused) {
|
||||
throw new ConnectException("connect refused");
|
||||
} else {
|
||||
throw new SocketTimeoutException("connect time out");
|
||||
}
|
||||
}
|
||||
|
||||
private volatile boolean connecting = false;
|
||||
|
||||
public boolean isConnecting() {
|
||||
return connecting;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void bind(InetAddress host, int port) throws IOException {
|
||||
if (host.equals(Inet4Address.getByName("0.0.0.0"))) {
|
||||
host = Inet6Address.getByName("::0");
|
||||
}
|
||||
if (!(host instanceof Inet6Address)) {
|
||||
throw new IllegalArgumentException("invalid address type, KLALB socket can only use IPV6 virtualaddress");
|
||||
}
|
||||
if ((!host.isAnyLocalAddress()) && (!host.equals(controller.getSelf()))) {
|
||||
throw new BindException("must bind to self");
|
||||
}
|
||||
bindaddr = (Inet6Address) host;
|
||||
this.localport = controller.bind(this, port);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void listen(int backlog) throws IOException {
|
||||
backlogQueue = new ArrayBlockingQueue<>(backlog);
|
||||
}
|
||||
|
||||
private Map<InetSocketAddress, KLALBVirtualSocketImpl> accepts = new ConcurrentHashMap<>();
|
||||
|
||||
public Map<InetSocketAddress, KLALBVirtualSocketImpl> getAccepts() {
|
||||
return accepts;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void accept(SocketImpl s) throws IOException {
|
||||
try {
|
||||
KLALBVirtualSocketImpl kvsi = (KLALBVirtualSocketImpl) s;
|
||||
InetSocketAddress isa = backlogQueue.take();
|
||||
|
||||
kvsi.port = isa.getPort();
|
||||
kvsi.address = isa.getAddress();
|
||||
kvsi.localport = localport;
|
||||
kvsi.bindaddr = bindaddr;
|
||||
accepts.put(isa, kvsi);
|
||||
kvsi.setCloseListener((x) -> {
|
||||
accepts.remove(isa);
|
||||
});
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
|
||||
private Consumer<KLALBVirtualSocketImpl> acceptedSocketCloseListener;
|
||||
|
||||
private void setCloseListener(Consumer<KLALBVirtualSocketImpl> lsr) {
|
||||
this.acceptedSocketCloseListener = lsr;
|
||||
}
|
||||
|
||||
private KVSIInputStream vin;
|
||||
private KVSIOutputStream vout;
|
||||
|
||||
private class KVSIInputStream extends InputStream {
|
||||
private DATATPacket dtp = null;
|
||||
private int count = 0;
|
||||
|
||||
private boolean shutdown=false;
|
||||
@Override
|
||||
public int read() throws IOException {
|
||||
if(shutdown)
|
||||
return -1;
|
||||
if (dtp == null || (count >= dtp.getData().length && dtp.getData().length != 0)) {
|
||||
count = 0;
|
||||
while (true) {
|
||||
if (isClosed())
|
||||
throw new SocketException("Socket is closed");
|
||||
DATATPacket dtp2 = sendDeque.poll();
|
||||
if (dtp2 != null) {
|
||||
dtp = dtp2;
|
||||
if(countInputBytes() < inputchachesize) {
|
||||
controller.sendPacketToAddress((Inet6Address) address, new ACKTPacket(dtp.getDport(), dtp.getSport(), dtp2.getNumber(),
|
||||
true), 32768);
|
||||
}
|
||||
break;
|
||||
}
|
||||
try {
|
||||
Thread.sleep(1);
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
}
|
||||
}
|
||||
if (dtp.getData().length == 0) {
|
||||
return -1;
|
||||
} else
|
||||
return dtp.getData()[count++] & 0xff;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int read(byte[] b, int off, int len) throws IOException {
|
||||
if (b == null) {
|
||||
throw new NullPointerException();
|
||||
} else if (off < 0 || len < 0 || len > b.length - off) {
|
||||
throw new IndexOutOfBoundsException();
|
||||
} else if (len == 0) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
len = Math.min(len, available());
|
||||
|
||||
int c = read();
|
||||
if (c == -1) {
|
||||
return -1;
|
||||
}
|
||||
b[off] = (byte) c;
|
||||
|
||||
int i = 1;
|
||||
try {
|
||||
for (; i < len; i++) {
|
||||
c = read();
|
||||
if (c == -1) {
|
||||
break;
|
||||
}
|
||||
b[off + i] = (byte) c;
|
||||
}
|
||||
} catch (IOException ee) {
|
||||
}
|
||||
return i;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
shutdown=true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int available() throws IOException {
|
||||
AtomicInteger i = new AtomicInteger(0);
|
||||
sendDeque.forEach((V) -> {
|
||||
i.addAndGet(V.getData().length);
|
||||
});
|
||||
if (dtp != null)
|
||||
i.addAndGet(dtp.getData().length - count);
|
||||
return i.get();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private long outputcount = 0;
|
||||
|
||||
private class KVSIOutputStream extends OutputStream {
|
||||
|
||||
private ByteArrayOutputStream bos = new ByteArrayOutputStream(65535);
|
||||
|
||||
@Override
|
||||
public void write(int b) throws IOException {
|
||||
if (isClosed())
|
||||
throw new SocketException("Socket is closed");
|
||||
bos.write(b);
|
||||
if (bos.size() >= 65535) {
|
||||
flush();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void flush() throws IOException {
|
||||
if (bos.size() > 0) {
|
||||
try {
|
||||
while (!avaliable) {
|
||||
Thread.sleep(1);
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
e.printStackTrace();
|
||||
}
|
||||
try {
|
||||
while(countOutputBytes()>outputchachesize) {
|
||||
Thread.sleep(1);
|
||||
}
|
||||
} catch (InterruptedException e) {
|
||||
// TODO 自动生成的 catch 块
|
||||
e.printStackTrace();
|
||||
}
|
||||
SendTask x=new SendTask(controller, (Inet6Address) address, new DATATPacket(localport, port, outputcount++, bos.toByteArray()), 5,5);
|
||||
x.addToTimer(1000);
|
||||
sendlist.add(x);
|
||||
/*controller.sendPacketToAddress((Inet6Address) address,
|
||||
new DATATPacket(localport, port, outputcount++, bos.toByteArray()), 5);*/
|
||||
}
|
||||
bos.reset();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() throws IOException {
|
||||
flush();
|
||||
SendTask x=new SendTask(controller, (Inet6Address) address, new DATATPacket(localport, port, outputcount++, new byte[0]), 5,5);
|
||||
x.addToTimer(1000);
|
||||
sendlist.add(x);
|
||||
/*controller.sendPacketToAddress((Inet6Address) address,
|
||||
new DATATPacket(localport, port, outputcount++, new byte[0]), 5);*/
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
protected InputStream getInputStream() throws IOException {
|
||||
if (vin == null) {
|
||||
vin = new KVSIInputStream();
|
||||
}
|
||||
return vin;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected OutputStream getOutputStream() throws IOException {
|
||||
if (vout == null) {
|
||||
vout = new KVSIOutputStream();
|
||||
}
|
||||
return vout;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected int available() throws IOException {
|
||||
return getInputStream().available();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void shutdownInput() throws IOException {
|
||||
getInputStream().close();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void shutdownOutput() throws IOException {
|
||||
getOutputStream().close();
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void setPerformancePreferences(int connectionTime, int latency, int bandwidth) {
|
||||
// TODO 自动生成的方法存根
|
||||
super.setPerformancePreferences(connectionTime, latency, bandwidth);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void close() throws IOException {
|
||||
if (!isListening() && !isClosed()) {
|
||||
closed = true;
|
||||
controller.sendPacketToAddress((Inet6Address) super.address, new RSTPacket(super.localport, super.port),
|
||||
65537);
|
||||
}
|
||||
if (acceptedSocketCloseListener != null)
|
||||
acceptedSocketCloseListener.accept(this);
|
||||
else
|
||||
controller.unbind(this);
|
||||
}
|
||||
|
||||
private boolean closed;
|
||||
|
||||
private boolean isClosed() {
|
||||
return closed;
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void sendUrgentData(int data) throws IOException {
|
||||
throw new UnsupportedOperationException("urgent data unspuuorted");
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
protected int getLocalPort() {
|
||||
// TODO 自动生成的方法存根
|
||||
return super.getLocalPort();
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user