fix ForwardCommand

Author: wangqifox

pom.xml

@@ -28,7 +28,7 @@
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-all</artifactId>
-            <version>4.1.25.Final</version>
+            <version>4.1.28.Final</version>
         </dependency>
         <dependency>
             <groupId>com.netflix.hystrix</groupId>

src/main/java/love/wangqi/codec/DefaultHttpRequestBuilder.java

@@ -28,28 +28,13 @@ public class DefaultHttpRequestBuilder implements HttpRequestBuilder {
     private static final Logger logger = LoggerFactory.getLogger(DefaultHttpRequestBuilder.class);
 
     private RouteMapper routeMapper;
-    private FullHttpRequest originRequest;
     private HttpRequest newRequest;
     private HttpPostRequestEncoder newBodyRequestEncoder;
 
     protected HttpRequestDecomposer httpRequestDecomposer;
 
     HttpDataFactory factory = new DefaultHttpDataFactory(DefaultHttpDataFactory.MINSIZE);
 
-    public class RequestHolder {
-        public Route route;
-        public URL url;
-        public HttpRequest request;
-        public HttpPostRequestEncoder bodyRequestEncoder;
-
-        public RequestHolder(Route route, URL url, HttpRequest request, HttpPostRequestEncoder bodyRequestEncoder) {
-            this.route = route;
-            this.url = url;
-            this.request = request;
-            this.bodyRequestEncoder = bodyRequestEncoder;
-        }
-    }
-
     public DefaultHttpRequestBuilder() {
     }
 
@@ -60,25 +45,19 @@ public DefaultHttpRequestBuilder setRouteMapper(RouteMapper routeMapper) {
     }
 
     @Override
-    public DefaultHttpRequestBuilder setOriginRequest(FullHttpRequest originRequest) {
-        this.originRequest = originRequest;
-        this.httpRequestDecomposer = new HttpRequestDecomposer(originRequest);
-        return this;
-    }
-
-    @Override
-    public Route getRoute() {
-        return routeMapper.getRoute(this.originRequest);
+    public Route getRoute(FullHttpRequest originRequest) {
+        return routeMapper.getRoute(originRequest);
     }
 
     @Override
-    public RequestHolder build() throws Exception {
-        Route route = getRoute();
+    public RequestHolder build(FullHttpRequest originRequest) throws Exception {
+        httpRequestDecomposer = new HttpRequestDecomposer(originRequest);
+        Route route = getRoute(originRequest);
         if (route == null) {
             throw new NoRouteException();
         }
         URL url  = route.getMapUrl();
-        logger.info(url.toString());
+//        logger.info(url.toString());
 
         // 请求路径
         QueryStringEncoder queryStringEncoder = new QueryStringEncoder(url.getPath());

src/main/java/love/wangqi/codec/HttpRequestBuilder.java

@@ -18,23 +18,16 @@ public interface HttpRequestBuilder {
      */
     DefaultHttpRequestBuilder setRouteMapper(RouteMapper routeMapper);
 
-    /**
-     * 设置原始请求
-     * @param originRequest
-     * @return
-     */
-    DefaultHttpRequestBuilder setOriginRequest(FullHttpRequest originRequest);
-
     /**
      * 生成新的请求
      * @return
      * @throws Exception
      */
-    DefaultHttpRequestBuilder.RequestHolder build() throws Exception;
+    RequestHolder build(FullHttpRequest originRequest) throws Exception;
 
     /**
      * 获取路由
      * @return
      */
-    Route getRoute();
+    Route getRoute(FullHttpRequest originRequest);
 }

src/main/java/love/wangqi/codec/RequestHolder.java

@@ -0,0 +1,26 @@
+package love.wangqi.codec;
+
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.multipart.HttpPostRequestEncoder;
+import love.wangqi.route.Route;
+
+import java.net.URL;
+
+/**
+ * @author: wangqi
+ * @description:
+ * @date: Created in 2018/7/28 22:37
+ */
+public class RequestHolder {
+    public Route route;
+    public URL url;
+    public HttpRequest request;
+    public HttpPostRequestEncoder bodyRequestEncoder;
+
+    public RequestHolder(Route route, URL url, HttpRequest request, HttpPostRequestEncoder bodyRequestEncoder) {
+        this.route = route;
+        this.url = url;
+        this.request = request;
+        this.bodyRequestEncoder = bodyRequestEncoder;
+    }
+}

src/main/java/love/wangqi/exception/handler/DefaultExceptionHandler.java

@@ -10,6 +10,7 @@
 import love.wangqi.exception.TimeoutException;
 
 import java.net.ConnectException;
+import java.util.concurrent.RejectedExecutionException;
 
 /**
  * @author: wangqi
@@ -32,6 +33,10 @@ public ExceptionResponse getExceptionResponse(Exception exception) {
             exceptionResponse.setStatus(HttpResponseStatus.REQUEST_TIMEOUT);
             exceptionResponse.setContentType("text/plain");
             exceptionResponse.setContent("request timeout");
+        } else if (exception instanceof RejectedExecutionException) {
+            exceptionResponse.setStatus(HttpResponseStatus.TOO_MANY_REQUESTS);
+            exceptionResponse.setContentType("text/plain");
+            exceptionResponse.setContent("too many requests");
         } else {
             exceptionResponse.setStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
             exceptionResponse.setContentType("text/plain");
@@ -43,11 +48,11 @@ public ExceptionResponse getExceptionResponse(Exception exception) {
     @Override
     public void send(ChannelHandlerContext ctx, ExceptionResponse exceptionResponse) {
         String content = exceptionResponse.getContent();
-        FullHttpResponse response;
-        response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, exceptionResponse.getStatus());
+        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, exceptionResponse.getStatus());
         if (content != null) {
             response.headers().set("X-Ca-Error-Message", content);
         }
-        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+        ctx.channel().writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+        ctx.channel().close();
     }
 }

src/main/java/love/wangqi/handler/BackendFilter.java

@@ -1,6 +1,6 @@
 package love.wangqi.handler;
 
-import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelPipeline;
 import io.netty.channel.socket.SocketChannel;
@@ -17,29 +17,23 @@
  */
 public class BackendFilter extends ChannelInitializer<SocketChannel> {
     private final SslContext sslCtx;
-    private final Channel inboundChannel;
+    private final ChannelHandlerContext ctx;
 
-    public BackendFilter(SslContext sslCtx, Channel inboundChannel) {
+    public BackendFilter(SslContext sslCtx, ChannelHandlerContext ctx) {
         this.sslCtx = sslCtx;
-        this.inboundChannel = inboundChannel;
+        this.ctx = ctx;
     }
 
     @Override
     protected void initChannel(SocketChannel ch) throws Exception {
         ChannelPipeline pipeline = ch.pipeline();
-
         if (sslCtx != null) {
             pipeline.addLast("ssl", sslCtx.newHandler(ch.alloc()));
         }
-
         pipeline.addLast("codec", new HttpClientCodec());
-
         pipeline.addLast("inflater", new HttpContentDecompressor());
-
-        pipeline.addLast(new HttpObjectAggregator(1048576));
-
+        pipeline.addLast(new HttpObjectAggregator(1024 * 1024 * 64));
         pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());
-
-        pipeline.addLast("handler", new BackendHandler(inboundChannel));
+        pipeline.addLast("handler", new BackendHandler(ctx));
     }
 }

src/main/java/love/wangqi/handler/BackendHandler.java

@@ -1,9 +1,9 @@
 package love.wangqi.handler;
 
-import io.netty.channel.Channel;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.handler.codec.http.FullHttpResponse;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -15,15 +15,15 @@
 public class BackendHandler extends ChannelInboundHandlerAdapter {
     private Logger logger = LoggerFactory.getLogger(BackendHandler.class);
 
-    private Channel inboundChannel;
+    private ChannelHandlerContext ctx;
 
-    BackendHandler(Channel inboundChannel) {
-        this.inboundChannel = inboundChannel;
+    BackendHandler(ChannelHandlerContext ctx) {
+        this.ctx = ctx;
     }
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
-        inboundChannel.writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE);
+        this.ctx.channel().writeAndFlush(msg).addListener(ChannelFutureListener.CLOSE);
         ctx.channel().close();
     }
 }

src/main/java/love/wangqi/handler/FrontHandler.java

@@ -2,9 +2,10 @@
 
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
-import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.*;
 import love.wangqi.codec.DefaultHttpRequestBuilder;
 import love.wangqi.codec.HttpRequestBuilder;
+import love.wangqi.codec.RequestHolder;
 import love.wangqi.filter.HttpRequestFilter;
 import love.wangqi.handler.command.ForwardCommand;
 import love.wangqi.server.GatewayServer;
@@ -33,24 +34,19 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
                 httpRequestFilter.filter(GatewayServer.config, ctx, httpRequest);
             }
 
-            HttpRequestBuilder httpRequestBuilder = GatewayServer.config.getHttpRequestBuilder()
-                    .setRouteMapper(GatewayServer.config.getRouteMapper())
-                    .setOriginRequest(httpRequest);
+            HttpRequestBuilder httpRequestBuilder = new DefaultHttpRequestBuilder()
+                    .setRouteMapper(GatewayServer.config.getRouteMapper());
+
+            RequestHolder requestHolder = httpRequestBuilder.build(httpRequest);
 
-            DefaultHttpRequestBuilder.RequestHolder requestHolder = httpRequestBuilder.build();
             ForwardCommand forwardCommand = new ForwardCommand(ctx, requestHolder);
             forwardCommand.queue();
         } catch (Exception e) {
             logger.error(e.toString());
             GatewayServer.config.getExceptionHandler().handle(ctx, e);
-        } finally {
+        }
+        finally {
             httpRequest.release();
         }
     }
-
-    @Override
-    public void channelActive(ChannelHandlerContext ctx) throws Exception {
-        logger.info("连接的客户端地址：{}", ctx.channel().remoteAddress());
-        super.channelActive(ctx);
-    }
 }

src/main/java/love/wangqi/handler/command/ForwardCommand.java

@@ -13,7 +13,7 @@
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
-import love.wangqi.codec.DefaultHttpRequestBuilder;
+import love.wangqi.codec.RequestHolder;
 import love.wangqi.exception.TimeoutException;
 import love.wangqi.handler.BackendFilter;
 import love.wangqi.server.GatewayServer;
@@ -27,9 +27,9 @@
  */
 public class ForwardCommand extends HystrixCommand<Void> {
     private ChannelHandlerContext ctx;
-    private DefaultHttpRequestBuilder.RequestHolder requestHolder;
+    private RequestHolder requestHolder;
 
-    public ForwardCommand(ChannelHandlerContext ctx, DefaultHttpRequestBuilder.RequestHolder requestHolder) {
+    public ForwardCommand(ChannelHandlerContext ctx, RequestHolder requestHolder) {
         super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("ForwardCommandGroup"))
                 .andCommandKey(HystrixCommandKey.Factory.asKey("ForwardCommand"))
                 .andThreadPoolKey(HystrixThreadPoolKey.Factory.asKey("threadPool-" + requestHolder.route.getId()))
@@ -46,15 +46,14 @@ public ForwardCommand(ChannelHandlerContext ctx, DefaultHttpRequestBuilder.Reque
                         .withFallbackIsolationSemaphoreMaxConcurrentRequests(100)
                 )
         );
-
         this.ctx = ctx;
         this.requestHolder = requestHolder;
     }
 
 
     @Override
     protected Void run() throws Exception {
-        forward(ctx.channel(), requestHolder.url, requestHolder.request, requestHolder.bodyRequestEncoder);
+        forward(ctx, requestHolder.url, requestHolder.request, requestHolder.bodyRequestEncoder);
         return null;
     }
 
@@ -99,7 +98,7 @@ private UrlMetadata getProtocol(URL url) {
         return new UrlMetadata(protocol, host, port);
     }
 
-    private void forward(Channel inboundChannel, URL url, HttpRequest request, HttpPostRequestEncoder bodyRequestEncoder) throws Exception {
+    private void forward(ChannelHandlerContext ctx, URL url, HttpRequest request, HttpPostRequestEncoder bodyRequestEncoder) throws Exception {
         UrlMetadata urlMetadata = getProtocol(url);
         // Configure SSL context if necessary.
         final boolean ssl = "https".equalsIgnoreCase(urlMetadata.protocol);
@@ -111,12 +110,12 @@ private void forward(Channel inboundChannel, URL url, HttpRequest request, HttpP
             sslCtx = null;
         }
 
-        EventLoopGroup group = new NioEventLoopGroup();
+        EventLoopGroup group = new NioEventLoopGroup(1);
         try {
             Bootstrap b = new Bootstrap();
             b.group(group)
                     .channel(NioSocketChannel.class)
-                    .handler(new BackendFilter(sslCtx, inboundChannel));
+                    .handler(new BackendFilter(sslCtx, ctx));
 
             Channel ch = b.connect(urlMetadata.host, urlMetadata.port).sync().channel();