src/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/com/wentch/redkale/net/socks/NodeSocksServer.java

@@ -16,6 +16,7 @@ import java.net.*;
 import java.util.logging.*;
 
 /**
+ * < server protocol="SOCKS" host="0.0.0.0" port="1080" bindaddr="外网IP"> < /server>
  *
  * @author zhangjx
  */
@@ -58,16 +59,9 @@ public class NodeSocksServer extends NodeServer {
             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(format(servlet.getRequestid())).append(LINE_SEPARATOR);
+            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());
     }
 
-    private static String format(short value) {
-        String str = Integer.toHexString(value);
-        if (str.length() == 1) return "0x000" + str;
-        if (str.length() == 2) return "0x00" + str;
-        if (str.length() == 3) return "0x0" + str;
-        return "0x" + str;
-    }
 }

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

@@ -8,38 +8,32 @@ package com.wentch.redkale.net.socks;
 import com.wentch.redkale.net.*;
 import com.wentch.redkale.util.*;
 import java.io.*;
-import java.util.*;
 
 /**
  *
  * @author zhangjx
  */
-public class SocksPrepareServlet  extends PrepareServlet<SocksRequest, SocksResponse> {
+public final class SocksPrepareServlet extends PrepareServlet<SocksRequest, SocksResponse> {
 
-    private final HashMap<Short, SocksServlet> servletmaps = new HashMap<>();
+    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 addSocksServlet(SocksServlet servlet, AnyValue conf) {
+    public void setSocksServlet(SocksServlet servlet, AnyValue conf) {
         servlet.conf = conf;
-        this.servletmaps.put(servlet.getRequestid(), servlet);
+        if (servlet != null) this.servlet = servlet;
     }
 
-    // 28.[00,03,00,08,  21,12,a4,42,45,6f,4e,77,4e,47,71,55,32,37,77,39,    00,19,00,04,11,00,00,00]
+    // 
     @Override
     public void execute(SocksRequest request, SocksResponse response) throws IOException {
-        SocksServlet servlet = servletmaps.get(request.getRequestid());
-        if (servlet != null) {
-            servlet.execute(request, response);
-        } else {
-            response.finish();
-        }
+        servlet.execute(request, response);
     }
 
 }
-

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

@@ -22,12 +22,14 @@ public class SocksRequest extends Request {
 
     @Override
     protected int readHeader(ByteBuffer buffer) {
-        throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
+        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) {
-        throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
     }
 
     @Override
@@ -36,6 +38,7 @@ public class SocksRequest extends Request {
 
     @Override
     protected void recycle() {
+        this.requestid = 0;
         super.recycle();
     }
 

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

@@ -18,9 +18,13 @@ 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/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/com/wentch/redkale/net/socks/SocksServer.java

@@ -31,7 +31,7 @@ public final class SocksServer extends Server {
     }
 
     public void addSocksServlet(SocksServlet servlet, AnyValue conf) {
-        ((SocksPrepareServlet) this.prepare).addSocksServlet(servlet, conf);
+        ((SocksPrepareServlet) this.prepare).setSocksServlet(servlet, conf);
     }
 
     @Override

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

@@ -16,8 +16,6 @@ public abstract class SocksServlet implements Servlet<SocksRequest, SocksRespons
 
     AnyValue conf; //当前Servlet的配置
 
-    public abstract short getRequestid();
-
     @Override
     public final boolean equals(Object obj) {
         return obj != null && obj.getClass() == this.getClass();