后端请求异步化

Author: wangqifox

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

@@ -39,7 +39,7 @@ public DefaultHttpRequestBuilder() {
     }
 
     @Override
-    public DefaultHttpRequestBuilder setRouteMapper(RouteMapper routeMapper) {
+    public HttpRequestBuilder setRouteMapper(RouteMapper routeMapper) {
         this.routeMapper = routeMapper;
         return this;
     }

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

@@ -16,17 +16,19 @@ public interface HttpRequestBuilder {
      * @param routeMapper
      * @return
      */
-    DefaultHttpRequestBuilder setRouteMapper(RouteMapper routeMapper);
+    HttpRequestBuilder setRouteMapper(RouteMapper routeMapper);
 
     /**
      * 生成新的请求
+     * @param originRequest
      * @return
      * @throws Exception
      */
     RequestHolder build(FullHttpRequest originRequest) throws Exception;
 
     /**
      * 获取路由
+     * @param originRequest
      * @return
      */
     Route getRoute(FullHttpRequest originRequest);

src/main/java/love/wangqi/config/GatewayConfig.java

@@ -4,6 +4,7 @@
 import io.netty.channel.ChannelInboundHandler;
 import io.netty.channel.ChannelOutboundHandler;
 import love.wangqi.codec.HttpRequestBuilder;
+import love.wangqi.core.ResponseHandler;
 import love.wangqi.exception.handler.ExceptionHandler;
 import love.wangqi.filter.FilterRegistry;
 import love.wangqi.filter.GatewayFilter;
@@ -80,6 +81,10 @@ public List<GatewayFilter> getFiltersByType(String filterType) {
      * 异常处理器
      */
     private ExceptionHandler exceptionHandler;
+    /**
+     * 响应处理器
+     */
+    private ResponseHandler responseHandler;
 
 
     private GatewayConfig() {
@@ -151,4 +156,12 @@ public ExceptionHandler getExceptionHandler() {
     public void setExceptionHandler(ExceptionHandler exceptionHandler) {
         this.exceptionHandler = exceptionHandler;
     }
+
+    public ResponseHandler getResponseHandler() {
+        return responseHandler;
+    }
+
+    public void setResponseHandler(ResponseHandler responseHandler) {
+        this.responseHandler = responseHandler;
+    }
 }

src/main/java/love/wangqi/context/HttpRequestContext.java

@@ -2,6 +2,7 @@
 
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.FullHttpResponse;
 import io.netty.handler.codec.http.HttpRequest;
 
 import java.util.concurrent.ConcurrentHashMap;
@@ -30,7 +31,7 @@ public static HttpRequestContext getInstance() {
     }
 
     public void setRequestChannel(HttpRequest httpRequest, Channel channel) {
-        requestChannel.putIfAbsent(httpRequest, channel);
+        requestChannel.put(httpRequest, channel);
         set(channel, RequestConstant.HTTPREQUEST, httpRequest);
     }
 
@@ -47,9 +48,9 @@ public void set(Channel channel, String key, Object value) {
         ConcurrentHashMap<String, Object> context = getContext(channel);
         if (context == null) {
             context = new ConcurrentHashMap<>();
-            channelContext.putIfAbsent(channel, context);
+            channelContext.put(channel, context);
         }
-        context.putIfAbsent(key, value);
+        context.put(key, value);
     }
 
     public <T> T get(HttpRequest httpRequest, String key) {
@@ -102,15 +103,15 @@ public void setException(Channel channel, Exception exception) {
         set(channel, RequestConstant.EXCEPTION, exception);
     }
 
-    public void setResponse(Channel channel, Object response) {
+    public void setResponse(Channel channel, FullHttpResponse response) {
         set(channel, RequestConstant.RESPONSE, response);
     }
 
-    public Object getResponse(Channel channel) {
+    public FullHttpResponse getResponse(Channel channel) {
         return get(channel, RequestConstant.RESPONSE);
     }
 
-    public Object getResponse(HttpRequest httpRequest) {
+    public FullHttpResponse getResponse(HttpRequest httpRequest) {
         return get(httpRequest, RequestConstant.RESPONSE);
     }
 }

src/main/java/love/wangqi/context/RequestConstant.java

@@ -11,4 +11,5 @@ public class RequestConstant {
     public static final String EXCEPTION = "exception";
     public static final String STATUSCODE = "statusCode";
     public static final String RESPONSE = "response";
+    public static final String KEEPALIVE = "keepAlive";
 }

src/main/java/love/wangqi/core/ResponseHandler.java

@@ -0,0 +1,39 @@
+package love.wangqi.core;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpHeaderNames;
+import io.netty.handler.codec.http.HttpUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import love.wangqi.config.GatewayConfig;
+import love.wangqi.context.HttpRequestContext;
+import love.wangqi.context.RequestConstant;
+
+/**
+ * @author: wangqi
+ * @description:
+ * @date: Created in 2018/9/28 下午6:14
+ */
+public class ResponseHandler {
+    private Logger logger = LoggerFactory.getLogger(ResponseHandler.class);
+
+    private GatewayConfig config = GatewayConfig.getInstance();
+    private HttpRequestContext httpRequestContext = HttpRequestContext.getInstance();
+
+    public void send(Channel channel, FullHttpResponse response) {
+        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
+        Boolean keepAlive = httpRequestContext.get(channel, RequestConstant.KEEPALIVE);
+        HttpUtil.setKeepAlive(response, keepAlive == null ? false : keepAlive);
+        channel.writeAndFlush(response).addListener(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) throws Exception {
+                if (config.getChannelCloseFutureListener() != null) {
+                    config.getChannelCloseFutureListener().operationComplete(channel, future);
+                }
+            }
+        });
+    }
+}

src/main/java/love/wangqi/exception/GatewayTimeoutException.java

@@ -10,6 +10,6 @@
 public class GatewayTimeoutException extends GatewayException {
 
     public GatewayTimeoutException() {
-        super(HttpResponseStatus.REQUEST_TIMEOUT, "timeout");
+        super(HttpResponseStatus.REQUEST_TIMEOUT, "request timeout");
     }
 }

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

@@ -1,11 +1,8 @@
 package love.wangqi.exception.handler;
 
-import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandlerContext;
-import io.netty.handler.codec.http.DefaultFullHttpResponse;
-import io.netty.handler.codec.http.FullHttpResponse;
-import io.netty.handler.codec.http.HttpResponseStatus;
-import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.codec.http.*;
+import love.wangqi.config.GatewayConfig;
 import love.wangqi.exception.GatewayException;
 
 import java.net.ConnectException;
@@ -17,6 +14,8 @@
  * @date: Created in 2018/6/5 下午6:16
  */
 public class DefaultExceptionHandler extends AbstractExceptionHandler {
+    private GatewayConfig config = GatewayConfig.getInstance();
+
     @Override
     public ExceptionResponse getExceptionResponse(Exception exception) {
         ExceptionResponse exceptionResponse = new ExceptionResponse();
@@ -45,10 +44,10 @@ public ExceptionResponse getExceptionResponse(Exception exception) {
     public void send(ChannelHandlerContext ctx, ExceptionResponse exceptionResponse) {
         String content = exceptionResponse.getContent();
         FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, exceptionResponse.getStatus());
+        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
         if (content != null) {
             response.headers().set("X-Ca-Error-Message", content);
         }
-        ctx.channel().writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
-        ctx.channel().close();
+        config.getResponseHandler().send(ctx.channel(), response);
     }
 }

src/main/java/love/wangqi/filter/SendForwardFilter.java

@@ -2,7 +2,6 @@
 
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.http.FullHttpRequest;
-import love.wangqi.codec.DefaultHttpRequestBuilder;
 import love.wangqi.codec.HttpRequestBuilder;
 import love.wangqi.codec.RequestHolder;
 import love.wangqi.config.GatewayConfig;
@@ -30,7 +29,7 @@ public int filterOrder() {
 
     @Override
     public void filter(FullHttpRequest httpRequest) throws Exception {
-        HttpRequestBuilder httpRequestBuilder = new DefaultHttpRequestBuilder()
+        HttpRequestBuilder httpRequestBuilder = config.getHttpRequestBuilder()
                 .setRouteMapper(config.getRouteMapper());
 
         RequestHolder requestHolder = httpRequestBuilder.build(httpRequest);

src/main/java/love/wangqi/filter/SendResponseFilter.java

@@ -1,8 +1,12 @@
 package love.wangqi.filter;
 
 import io.netty.channel.Channel;
-import io.netty.channel.ChannelFutureListener;
 import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import love.wangqi.handler.GatewayRunner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import love.wangqi.config.GatewayConfig;
 import love.wangqi.context.HttpRequestContext;
 
 /**
@@ -11,8 +15,12 @@
  * @date: Created in 2018/9/5 下午2:16
  */
 public class SendResponseFilter extends GatewayFilter {
+    private static final Logger logger = LoggerFactory.getLogger(SendResponseFilter.class);
+
     private HttpRequestContext httpRequestContext = HttpRequestContext.getInstance();
 
+    private GatewayConfig config = GatewayConfig.getInstance();
+
     @Override
     public String filterType() {
         return "post";
@@ -28,8 +36,11 @@ public void filter(FullHttpRequest httpRequest) throws Exception {
         Channel channel = httpRequestContext.getChannel(httpRequest);
         Exception e = httpRequestContext.getException(httpRequest);
         if (channel != null && e == null) {
-            Object response = httpRequestContext.getResponse(httpRequest);
-            channel.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+            FullHttpResponse response = httpRequestContext.getResponse(httpRequest);
+//            logger.info("*** content {}", response.content().toString(Charset.defaultCharset()));
+            config.getResponseHandler().send(channel, response);
+        } else {
+            GatewayRunner.getInstance().errorAction((FullHttpRequest) httpRequestContext.getHttpRequest(channel));
         }
     }
 }