src-plugin/com/wentch/redkale/net/socks/DefaultSocksServlet.java

@@ -0,0 +1,56 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package com.wentch.redkale.net.socks;
+
+import com.wentch.redkale.net.*;
+import com.wentch.redkale.util.*;
+import java.io.*;
+import java.net.*;
+import java.nio.*;
+import java.util.logging.*;
+
+/**
+ *
+ * @author zhangjx
+ */
+@AutoLoad(false)
+public class DefaultSocksServlet extends SocksServlet {
+
+    private InetSocketAddress bindAddress;
+
+    private byte[] bindAddressBytes = new byte[0];
+
+    @Override
+    public void init(Context context, AnyValue config) {
+        if (config == null) {
+            this.bindAddress = new InetSocketAddress(Utility.localInetAddress(), context.getServerAddress().getPort());
+        } else {
+            this.bindAddress = new InetSocketAddress(config.getValue("bindaddr", Utility.localInetAddress().getHostAddress()), context.getServerAddress().getPort());
+        }
+        Logger logger = context.getLogger();
+        if (logger.isLoggable(Level.INFO)) logger.info("[" + Thread.currentThread().getName() + "] bindAddress = " + bindAddress);
+        ByteBuffer bb;
+        InetAddress addr = bindAddress.getAddress();
+        if (addr instanceof Inet6Address) {
+            bb = ByteBuffer.allocate(1 + 16 + 2);
+            bb.put((byte) 0x04);
+        } else {
+            bb = ByteBuffer.allocate(1 + 4 + 2);
+            bb.put((byte) 0x01);
+        }
+        bb.put(addr.getAddress());
+        bb.putChar((char) bindAddress.getPort());
+        bb.flip();
+        this.bindAddressBytes = bb.array();
+    }
+
+    @Override
+    public void execute(SocksRequest request, SocksResponse response) throws IOException {
+        response.getContext().submit(new SocksRunner(response.getContext(), response.removeChannel(), bindAddressBytes));
+        response.finish(true);
+    }
+
+}

src-plugin/com/wentch/redkale/net/socks/NodeSocksServer.java

@@ -0,0 +1,67 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package com.wentch.redkale.net.socks;
+
+import com.wentch.redkale.boot.ClassFilter.FilterEntry;
+import com.wentch.redkale.boot.*;
+import static com.wentch.redkale.boot.NodeServer.LINE_SEPARATOR;
+import com.wentch.redkale.net.*;
+import com.wentch.redkale.util.*;
+import com.wentch.redkale.util.AnyValue.DefaultAnyValue;
+import java.lang.reflect.*;
+import java.net.*;
+import java.util.logging.*;
+
+/**
+ * < server protocol="SOCKS" host="0.0.0.0" port="1080" bindaddr="外网IP"> < /server>
+ *
+ * @author zhangjx
+ */
+@NodeProtocol({"SOCKS"})
+public class NodeSocksServer extends NodeServer {
+
+    private final SocksServer socksServer;
+
+    public NodeSocksServer(Application application, AnyValue serconf) {
+        super(application, application.getResourceFactory().createChild(), createServer(application, serconf));
+        this.socksServer = (SocksServer) server;
+    }
+
+    private static Server createServer(Application application, AnyValue serconf) {
+        return new SocksServer(application.getStartTime(), application.getWatchFactory());
+    }
+
+    @Override
+    public InetSocketAddress getSocketAddress() {
+        return socksServer == null ? null : socksServer.getSocketAddress();
+    }
+
+    @Override
+    protected ClassFilter<Servlet> createServletClassFilter() {
+        return createClassFilter(null, null, SocksServlet.class, null, "servlets", "servlet");
+    }
+
+    @Override
+    protected void loadServlet(ClassFilter<? extends Servlet> servletFilter) throws Exception {
+        if (socksServer != null) loadSocksServlet(this.nodeConf.getAnyValue("servlets"), servletFilter);
+    }
+
+    protected void loadSocksServlet(final AnyValue conf, ClassFilter<? extends Servlet> filter) throws Exception {
+        final StringBuilder sb = logger.isLoggable(Level.FINE) ? new StringBuilder() : null;
+        final String threadName = "[" + Thread.currentThread().getName() + "] ";
+        for (FilterEntry<? extends Servlet> en : filter.getFilterEntrys()) {
+            Class<SocksServlet> clazz = (Class<SocksServlet>) en.getType();
+            if (Modifier.isAbstract(clazz.getModifiers())) continue;
+            final SocksServlet servlet = clazz.newInstance();
+            factory.inject(servlet);
+            DefaultAnyValue servletConf = (DefaultAnyValue) en.getProperty();
+            this.socksServer.addSocksServlet(servlet, servletConf);
+            if (sb != null) sb.append(threadName).append(" Loaded ").append(clazz.getName()).append(" --> ").append(servletConf).append(LINE_SEPARATOR);
+        }
+        if (sb != null && sb.length() > 0) logger.log(Level.FINE, sb.toString());
+    }
+
+}

src-plugin/com/wentch/redkale/net/socks/SocksContext.java

@@ -0,0 +1,29 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package com.wentch.redkale.net.socks;
+
+import com.wentch.redkale.net.*;
+import com.wentch.redkale.util.*;
+import com.wentch.redkale.watch.*;
+import java.net.*;
+import java.nio.*;
+import java.nio.charset.*;
+import java.util.concurrent.*;
+import java.util.logging.*;
+
+/**
+ *
+ * @author zhangjx
+ */
+public class SocksContext extends Context {
+
+    public SocksContext(long serverStartTime, Logger logger, ExecutorService executor, ObjectPool<ByteBuffer> bufferPool,
+            ObjectPool<Response> responsePool, int maxbody, Charset charset, InetSocketAddress address, PrepareServlet prepare,
+            WatchFactory watch, int readTimeoutSecond, int writeTimeoutSecond) {
+        super(serverStartTime, logger, executor, bufferPool, responsePool, maxbody, charset,
+                address, prepare, watch, readTimeoutSecond, writeTimeoutSecond);
+    }
+}

src-plugin/com/wentch/redkale/net/socks/SocksPrepareServlet.java

@@ -0,0 +1,39 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package com.wentch.redkale.net.socks;
+
+import com.wentch.redkale.net.*;
+import com.wentch.redkale.util.*;
+import java.io.*;
+
+/**
+ *
+ * @author zhangjx
+ */
+public final class SocksPrepareServlet extends PrepareServlet<SocksRequest, SocksResponse> {
+
+    private SocksServlet servlet = new DefaultSocksServlet();
+
+    public SocksPrepareServlet() {
+    }
+
+    @Override
+    public void init(Context context, AnyValue config) {
+        if (servlet != null) servlet.init(context, servlet.conf == null ? config : servlet.conf);
+    }
+
+    public void setSocksServlet(SocksServlet servlet, AnyValue conf) {
+        servlet.conf = conf;
+        if (servlet != null) this.servlet = servlet;
+    }
+
+    // 
+    @Override
+    public void execute(SocksRequest request, SocksResponse response) throws IOException {
+        servlet.execute(request, response);
+    }
+
+}

src-plugin/com/wentch/redkale/net/socks/SocksRequest.java

@@ -0,0 +1,53 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package com.wentch.redkale.net.socks;
+
+import com.wentch.redkale.net.*;
+import java.nio.*;
+
+/**
+ *
+ * @author zhangjx
+ */
+public class SocksRequest extends Request {
+
+    private short requestid;
+
+    protected SocksRequest(SocksContext context) {
+        super(context);
+    }
+
+    @Override
+    protected int readHeader(ByteBuffer buffer) {
+        if (buffer.get() != 0x05) return -1;
+        if (buffer.get() != 0x01) return -1;
+        if (buffer.get() != 0x00) return -1;
+        return 0;
+    }
+
+    @Override
+    protected void readBody(ByteBuffer buffer) {
+    }
+
+    @Override
+    protected void prepare() {
+    }
+
+    @Override
+    protected void recycle() {
+        this.requestid = 0;
+        super.recycle();
+    }
+
+    public short getRequestid() {
+        return requestid;
+    }
+
+    public void setRequestid(short requestid) {
+        this.requestid = requestid;
+    }
+
+}

src-plugin/com/wentch/redkale/net/socks/SocksResponse.java

@@ -0,0 +1,30 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package com.wentch.redkale.net.socks;
+
+import com.wentch.redkale.net.*;
+import com.wentch.redkale.util.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ *
+ * @author zhangjx
+ */
+public class SocksResponse extends Response<SocksRequest> {
+
+    protected SocksResponse(Context context, SocksRequest request) {
+        super(context, request);
+    }
+
+    public static ObjectPool<Response> createPool(AtomicLong creatCounter, AtomicLong cycleCounter, int max, Creator<Response> creator) {
+        return new ObjectPool<>(creatCounter, cycleCounter, max, creator, (x) -> ((SocksResponse) x).prepare(), (x) -> ((SocksResponse) x).recycle());
+    }
+
+    @Override
+    public AsyncConnection removeChannel() {
+        return super.removeChannel();
+    }
+}

src-plugin/com/wentch/redkale/net/socks/SocksRunner.java

@@ -0,0 +1,199 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package com.wentch.redkale.net.socks;
+
+import com.wentch.redkale.net.*;
+import java.io.*;
+import java.net.*;
+import java.nio.*;
+import java.nio.channels.*;
+import java.util.logging.*;
+
+/**
+ *
+ * @author zhangjx
+ */
+public class SocksRunner implements Runnable {
+
+    private final AsyncConnection channel;
+
+    private final Logger logger;
+
+    private final boolean finest;
+
+    private final Context context;
+
+    private final byte[] bindAddressBytes;
+
+    private ByteBuffer buffer;
+
+    protected boolean closed = false;
+
+    private InetSocketAddress remoteAddress;
+
+    private AsyncConnection remoteChannel;
+
+    public SocksRunner(Context context, AsyncConnection channel, final byte[] bindAddressBytes) {
+        this.context = context;
+        this.logger = context.getLogger();
+        this.finest = this.context.getLogger().isLoggable(Level.FINEST);
+        this.channel = channel;
+        this.buffer = context.pollBuffer();
+        this.bindAddressBytes = bindAddressBytes;
+    }
+
+    @Override
+    public void run() {
+        try {
+            ask();
+        } catch (Exception e) {
+            closeRunner(e);
+        }
+    }
+
+    private void ask() {
+        buffer.putChar((char) 0x0500);
+        buffer.flip();
+        this.channel.write(buffer, null, new CompletionHandler<Integer, Void>() {
+
+            @Override
+            public void completed(Integer result, Void attachment) {
+                connect();
+            }
+
+            @Override
+            public void failed(Throwable exc, Void attachment) {
+                closeRunner(exc);
+            }
+        });
+    }
+
+    private void connect() {
+        buffer.clear();
+        this.channel.read(buffer, null, new CompletionHandler<Integer, Void>() {
+
+            @Override
+            public void completed(Integer result, Void attachment) {
+                buffer.flip();
+                if (buffer.getChar() != 0x0501) {
+                    if (finest) logger.finest("connect header not 0x0501");
+                    closeRunner(null);
+                    return;
+                }
+                char addrtype = buffer.getChar(); //0x0001 - 4  ; 0x0003 - x ; 0x0004 - 16
+                try {
+                    byte[] bytes = new byte[(addrtype == 0x0003) ? (buffer.get() & 0xff) : (addrtype * 4)];
+                    buffer.get(bytes);
+                    remoteAddress = new InetSocketAddress((addrtype == 0x0003) ? InetAddress.getByName(new String(bytes)) : InetAddress.getByAddress(bytes), buffer.getChar());
+                } catch (UnknownHostException e) {
+                    throw new RuntimeException(e);
+                }
+                try {
+                    remoteChannel = AsyncConnection.create("TCP", remoteAddress, 6, 6);
+                    buffer.clear();
+                    buffer.putChar((char) 0x0500);
+                    buffer.put((byte) 0x00);  //rsv
+                    buffer.put(bindAddressBytes);
+                    buffer.flip();
+                    channel.write(buffer, null, new CompletionHandler<Integer, Void>() {
+
+                        @Override
+                        public void completed(Integer result, Void attachment) {
+                            stream();
+                        }
+
+                        @Override
+                        public void failed(Throwable exc, Void attachment) {
+                            closeRunner(exc);
+                        }
+                    });
+                } catch (IOException e) {
+                    buffer.clear();
+                    buffer.putChar((char) 0x0504);
+                    if (finest) logger.finest(remoteAddress + " remote connect error");
+                    channel.write(buffer, null, new CompletionHandler<Integer, Void>() {
+
+                        @Override
+                        public void completed(Integer result, Void attachment) {
+                            closeRunner(null);
+                        }
+
+                        @Override
+                        public void failed(Throwable exc, Void attachment) {
+                            closeRunner(exc);
+                        }
+                    });
+                }
+            }
+
+            @Override
+            public void failed(Throwable exc, Void attachment) {
+                closeRunner(exc);
+            }
+        });
+    }
+
+    private void stream() {
+        new StreamCompletionHandler(channel, remoteChannel).completed(0, null);
+        new StreamCompletionHandler(remoteChannel, channel).completed(0, null);
+    }
+
+    public synchronized void closeRunner(final Throwable e) {
+        if (closed) return;
+        closed = true;
+        try {
+            channel.close();
+        } catch (Throwable t) {
+        }
+        context.offerBuffer(buffer);
+        buffer = null;
+        if (e != null && finest) {
+            logger.log(Level.FINEST, "close socks channel by error", e);
+        }
+    }
+
+    private class StreamCompletionHandler implements CompletionHandler<Integer, Void> {
+
+        private final AsyncConnection conn1;
+
+        private final AsyncConnection conn2;
+
+        private final ByteBuffer rbuffer;
+
+        public StreamCompletionHandler(AsyncConnection conn1, AsyncConnection conn2) {
+            this.conn1 = conn1;
+            this.conn2 = conn2;
+            this.rbuffer = context.pollBuffer();
+        }
+
+        @Override
+        public void completed(Integer result0, Void v0) {
+            rbuffer.clear();
+            final CompletionHandler self = this;
+            conn1.read(rbuffer, null, new CompletionHandler<Integer, Void>() {
+
+                @Override
+                public void completed(Integer result, Void attachment) {
+                    rbuffer.flip();
+                    conn2.write(rbuffer, attachment, self);
+                }
+
+                @Override
+                public void failed(Throwable exc, Void attachment) {
+                    self.failed(exc, attachment);
+                }
+            });
+        }
+
+        @Override
+        public void failed(Throwable exc, Void v) {
+            context.offerBuffer(rbuffer);
+            conn1.dispose();
+            conn2.dispose();
+            if (finest) logger.log(Level.FINEST, "StreamCompletionHandler closed", exc);
+        }
+    }
+}

src-plugin/com/wentch/redkale/net/socks/SocksServer.java

@@ -0,0 +1,58 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package com.wentch.redkale.net.socks;
+
+import com.wentch.redkale.net.*;
+import com.wentch.redkale.util.*;
+import com.wentch.redkale.watch.*;
+import java.nio.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ *
+ * @author zhangjx
+ */
+public final class SocksServer extends Server {
+
+    public SocksServer() {
+        this(System.currentTimeMillis(), null);
+    }
+
+    public SocksServer(long serverStartTime, final WatchFactory watch) {
+        super(serverStartTime, "TCP", new SocksPrepareServlet(), watch);
+    }
+
+    @Override
+    public void init(AnyValue config) throws Exception {
+        super.init(config);
+    }
+
+    public void addSocksServlet(SocksServlet servlet, AnyValue conf) {
+        ((SocksPrepareServlet) this.prepare).setSocksServlet(servlet, conf);
+    }
+
+    @Override
+    @SuppressWarnings("unchecked")
+    protected Context createContext() {
+        final int port = this.address.getPort();
+        AtomicLong createBufferCounter = watch == null ? new AtomicLong() : watch.createWatchNumber("SOCKS_" + port + ".Buffer.creatCounter");
+        AtomicLong cycleBufferCounter = watch == null ? new AtomicLong() : watch.createWatchNumber("SOCKS_" + port + ".Buffer.cycleCounter");
+        int rcapacity = Math.max(this.capacity, 8 * 1024);
+        ObjectPool<ByteBuffer> bufferPool = new ObjectPool<>(createBufferCounter, cycleBufferCounter, this.bufferPoolSize,
+                (Object... params) -> ByteBuffer.allocateDirect(rcapacity), null, (e) -> {
+                    if (e == null || e.isReadOnly() || e.capacity() != rcapacity) return false;
+                    e.clear();
+                    return true;
+                });
+        AtomicLong createResponseCounter = watch == null ? new AtomicLong() : watch.createWatchNumber("SOCKS_" + port + ".Response.creatCounter");
+        AtomicLong cycleResponseCounter = watch == null ? new AtomicLong() : watch.createWatchNumber("SOCKS_" + port + ".Response.cycleCounter");
+        ObjectPool<Response> responsePool = SocksResponse.createPool(createResponseCounter, cycleResponseCounter, this.responsePoolSize, null);
+        SocksContext localcontext = new SocksContext(this.serverStartTime, this.logger, executor, bufferPool, responsePool,
+                this.maxbody, this.charset, this.address, this.prepare, this.watch, this.readTimeoutSecond, this.writeTimeoutSecond);
+        responsePool.setCreator((Object... params) -> new SocksResponse(localcontext, new SocksRequest(localcontext)));
+        return localcontext;
+    }
+}

src-plugin/com/wentch/redkale/net/socks/SocksServlet.java

@@ -0,0 +1,28 @@
+/*
+ * To change this license header, choose License Headers in Project Properties.
+ * To change this template file, choose Tools | Templates
+ * and open the template in the editor.
+ */
+package com.wentch.redkale.net.socks;
+
+import com.wentch.redkale.net.*;
+import com.wentch.redkale.util.*;
+
+/**
+ *
+ * @author zhangjx
+ */
+public abstract class SocksServlet implements Servlet<SocksRequest, SocksResponse> {
+
+    AnyValue conf; //当前Servlet的配置
+
+    @Override
+    public final boolean equals(Object obj) {
+        return obj != null && obj.getClass() == this.getClass();
+    }
+
+    @Override
+    public final int hashCode() {
+        return this.getClass().hashCode();
+    }
+}