增加了分层filter，filter之间线程隔离

Author: wangqifox

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

@@ -9,7 +9,7 @@
 import io.netty.handler.codec.http.multipart.InterfaceHttpData;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import love.wangqi.exception.NoRouteException;
+import love.wangqi.exception.GatewayNoRouteException;
 import love.wangqi.route.Route;
 import love.wangqi.route.RouteMapper;
 
@@ -54,7 +54,7 @@ public RequestHolder build(FullHttpRequest originRequest) throws Exception {
         httpRequestDecomposer = new HttpRequestDecomposer(originRequest);
         Route route = getRoute(originRequest);
         if (route == null) {
-            throw new NoRouteException();
+            throw new GatewayNoRouteException();
         }
         URL url  = route.getMapUrl();
 //        logger.info(url.toString());

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

@@ -5,19 +5,49 @@
 import io.netty.channel.ChannelOutboundHandler;
 import love.wangqi.codec.HttpRequestBuilder;
 import love.wangqi.exception.handler.ExceptionHandler;
-import love.wangqi.filter.HttpRequestFilter;
+import love.wangqi.filter.FilterRegistry;
+import love.wangqi.filter.GatewayFilter;
 import love.wangqi.listener.ChannelCloseFutureListener;
 import love.wangqi.route.RouteMapper;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
 
 /**
  * @author: wangqi
  * @description:
  * @date: Created in 2018/6/4 下午6:49
  */
 public class GatewayConfig {
+
+    final static GatewayConfig INSTANCE = new GatewayConfig();
+
+    private final ConcurrentHashMap<String, List<GatewayFilter>> hashFiltersByType = new ConcurrentHashMap<String, List<GatewayFilter>>();
+
+    public static GatewayConfig getInstance() {
+        return INSTANCE;
+    }
+
+    public List<GatewayFilter> getFiltersByType(String filterType) {
+        List<GatewayFilter> list = hashFiltersByType.get(filterType);
+        if (list != null) {
+            return list;
+        }
+
+        list = new ArrayList<>();
+
+        Collection<GatewayFilter> filters = FilterRegistry.instance().getAllFilters();
+        for (Iterator<GatewayFilter> iterator = filters.iterator(); iterator.hasNext(); ) {
+            GatewayFilter filter = iterator.next();
+            if (filter.filterType().equals(filterType)) {
+                list.add(filter);
+            }
+        }
+        Collections.sort(list);
+        hashFiltersByType.putIfAbsent(filterType, list);
+        return list;
+    }
+
     /**
      * 读入流量处理器
      */
@@ -46,21 +76,16 @@ public class GatewayConfig {
      * channel关闭监听器
      */
     private ChannelCloseFutureListener channelCloseFutureListener;
-    /**
-     * HttpRequest过滤器
-     */
-    private List<HttpRequestFilter> httpRequestFilterList;
     /**
      * 异常处理器
      */
     private ExceptionHandler exceptionHandler;
 
 
-    public GatewayConfig() {
+    private GatewayConfig() {
         channelInboundHandlerList = new ArrayList<>();
         channelOutboundHandlerList = new ArrayList<>();
         httpResponseHandlerList = new ArrayList<>();
-        httpRequestFilterList = new ArrayList<>();
     }
 
     public void addChannelInboundHandler(ChannelInboundHandler channelInboundHandler) {
@@ -119,14 +144,6 @@ public void setChannelCloseFutureListener(ChannelCloseFutureListener channelClos
         this.channelCloseFutureListener = channelCloseFutureListener;
     }
 
-    public void addHttpRequestFilter(HttpRequestFilter httpRequestFilter) {
-        httpRequestFilterList.add(httpRequestFilter);
-    }
-
-    public List<HttpRequestFilter> getHttpRequestFilterList() {
-        return httpRequestFilterList;
-    }
-
     public ExceptionHandler getExceptionHandler() {
         return exceptionHandler;
     }

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

@@ -0,0 +1,116 @@
+package love.wangqi.context;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author: wangqi
+ * @description:
+ * @date: Created in 2018/9/3 下午7:31
+ */
+public class HttpRequestContext {
+    /**
+     * HttpRequest相关联的数据
+     */
+    private static final ConcurrentHashMap<Channel, ConcurrentHashMap<String, Object>> channelContext = new ConcurrentHashMap<>();
+    /**
+     * Channel与HttpRequest的对应关系
+     */
+    private static final ConcurrentHashMap<HttpRequest, Channel> requestChannel = new ConcurrentHashMap<>();
+
+    private HttpRequestContext() {}
+
+    private static final HttpRequestContext INSTANCE = new HttpRequestContext();
+
+    public static HttpRequestContext getInstance() {
+        return INSTANCE;
+    }
+
+    public void setRequestChannel(HttpRequest httpRequest, Channel channel) {
+        requestChannel.putIfAbsent(httpRequest, channel);
+        set(channel, RequestConstant.HTTPREQUEST, httpRequest);
+    }
+
+    public ConcurrentHashMap<String, Object> getContext(Channel channel) {
+        return channelContext.get(channel);
+    }
+
+    public <T> T get(Channel channel, String key) {
+        ConcurrentHashMap<String, Object> context = getContext(channel);
+        return context == null ? null : (T) context.get(key);
+    }
+
+    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);
+        }
+        context.putIfAbsent(key, value);
+    }
+
+    public <T> T get(HttpRequest httpRequest, String key) {
+        Channel channel = requestChannel.get(httpRequest);
+        return channel == null ? null : get(channel, key);
+    }
+
+    public void set(HttpRequest httpRequest, String key, Object value) {
+        Channel channel = requestChannel.get(httpRequest);
+        set(channel, key, value);
+    }
+
+    public void remove(HttpRequest httpRequest) {
+        Channel channel = requestChannel.get(httpRequest);
+        ConcurrentHashMap<String, Object> map = channelContext.remove(channel);
+        requestChannel.remove(httpRequest);
+    }
+
+    public void remove(Channel channel) {
+        ConcurrentHashMap<String, Object> map = channelContext.remove(channel);
+        requestChannel.remove(map.get(RequestConstant.HTTPREQUEST));
+    }
+
+    public HttpRequest getHttpRequest(Channel channel) {
+        return get(channel, RequestConstant.HTTPREQUEST);
+    }
+
+    public Channel getChannel(HttpRequest httpRequest) {
+        return requestChannel.get(httpRequest);
+    }
+
+    public ChannelHandlerContext getChannelHandlerContext(HttpRequest httpRequest) {
+        return (ChannelHandlerContext) get(httpRequest, RequestConstant.CHANNELHANDLERCONTEXT);
+    }
+
+    public void setChannelHandlerContext(HttpRequest httpRequest, ChannelHandlerContext ctx) {
+        setRequestChannel(httpRequest, ctx.channel());
+        set(httpRequest, RequestConstant.CHANNELHANDLERCONTEXT, ctx);
+    }
+
+    public Exception getException(HttpRequest httpRequest) {
+        return (Exception) get(httpRequest, RequestConstant.EXCEPTION);
+    }
+
+    public void setException(HttpRequest httpRequest, Exception exception) {
+        set(httpRequest, RequestConstant.EXCEPTION, exception);
+    }
+
+    public void setException(Channel channel, Exception exception) {
+        set(channel, RequestConstant.EXCEPTION, exception);
+    }
+
+    public void setResponse(Channel channel, Object response) {
+        set(channel, RequestConstant.RESPONSE, response);
+    }
+
+    public Object getResponse(Channel channel) {
+        return get(channel, RequestConstant.RESPONSE);
+    }
+
+    public Object getResponse(HttpRequest httpRequest) {
+        return get(httpRequest, RequestConstant.RESPONSE);
+    }
+}

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

@@ -0,0 +1,14 @@
+package love.wangqi.context;
+
+/**
+ * @author: wangqi
+ * @description:
+ * @date: Created in 2018/9/5 上午11:02
+ */
+public class RequestConstant {
+    public static final String CHANNELHANDLERCONTEXT = "channelHandlerContext";
+    public static final String HTTPREQUEST = "httpRequest";
+    public static final String EXCEPTION = "exception";
+    public static final String STATUSCODE = "statusCode";
+    public static final String RESPONSE = "response";
+}

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

@@ -0,0 +1,28 @@
+package love.wangqi.exception;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * @author: wangqi
+ * @description:
+ * @date: Created in 2018/9/3 下午6:39
+ */
+public class GatewayException extends RuntimeException {
+    private HttpResponseStatus status;
+    private String message;
+
+    public GatewayException(HttpResponseStatus status, String message) {
+        super(message);
+        this.status = status;
+        this.message = message;
+    }
+
+    public HttpResponseStatus getStatus() {
+        return status;
+    }
+
+    @Override
+    public String getMessage() {
+        return message;
+    }
+}

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

@@ -0,0 +1,16 @@
+package love.wangqi.exception;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * @author: wangqi
+ * @description:
+ * @date: Created in 2018/6/5 下午7:26
+ */
+public class GatewayNoRouteException extends GatewayException {
+
+    public GatewayNoRouteException() {
+        super(HttpResponseStatus.NOT_FOUND, "no route found");
+    }
+
+}

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

@@ -0,0 +1,15 @@
+package love.wangqi.exception;
+
+import io.netty.handler.codec.http.HttpResponseStatus;
+
+/**
+ * @author: wangqi
+ * @description:
+ * @date: Created in 2018/7/28 09:37
+ */
+public class GatewayTimeoutException extends GatewayException {
+
+    public GatewayTimeoutException() {
+        super(HttpResponseStatus.REQUEST_TIMEOUT, "timeout");
+    }
+}

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

@@ -6,8 +6,7 @@
 import io.netty.handler.codec.http.FullHttpResponse;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.HttpVersion;
-import love.wangqi.exception.NoRouteException;
-import love.wangqi.exception.TimeoutException;
+import love.wangqi.exception.GatewayException;
 
 import java.net.ConnectException;
 import java.util.concurrent.RejectedExecutionException;
@@ -21,18 +20,15 @@ public class DefaultExceptionHandler extends AbstractExceptionHandler {
     @Override
     public ExceptionResponse getExceptionResponse(Exception exception) {
         ExceptionResponse exceptionResponse = new ExceptionResponse();
-        if (exception instanceof NoRouteException) {
-            exceptionResponse.setStatus(HttpResponseStatus.NOT_FOUND);
+        if (exception instanceof GatewayException) {
+            GatewayException gatewayException = (GatewayException) exception;
+            exceptionResponse.setStatus(gatewayException.getStatus());
             exceptionResponse.setContentType("text/plain");
-            exceptionResponse.setContent(exception.getMessage());
+            exceptionResponse.setContent(gatewayException.getMessage());
         } else if (exception instanceof ConnectException) {
             exceptionResponse.setStatus(HttpResponseStatus.NOT_FOUND);
             exceptionResponse.setContentType("text/plain");
             exceptionResponse.setContent("connect server refused");
-        } else if (exception instanceof TimeoutException) {
-            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");