Skip to content

Commit 924b3de

Browse files
wangqifoxwangqifox
committed
客户端加入连接池
1 parent 595d33c commit 924b3de

16 files changed

+359
-63
lines changed

‎src/main/java/love/wangqi/GatewayServerDemo.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
import io.netty.handler.codec.http.HttpMethod;
55
import love.wangqi.codec.DefaultHttpRequestBuilder;
66
import love.wangqi.config.GatewayConfig;
7-
import love.wangqi.core.DefaultChannelCloseFutureListener;
7+
import love.wangqi.core.DefaultChannelWriteFinishListener;
88
import love.wangqi.core.ResponseHandler;
99
import love.wangqi.exception.handler.DefaultExceptionHandler;
1010
import love.wangqi.route.AbstractRouteMapper;
@@ -50,7 +50,7 @@ protected List<Route> locateRouteList(Set<Long> ids) {
5050
config.setPort(8888);
5151
config.setHttpRequestBuilder(new DefaultHttpRequestBuilder());
5252
config.setRouteMapper(routeMapper);
53-
config.setChannelCloseFutureListener(new DefaultChannelCloseFutureListener());
53+
config.setChannelWriteFinishListener(new DefaultChannelWriteFinishListener());
5454
config.setResponseHandler(new ResponseHandler());
5555
config.setExceptionHandler(new DefaultExceptionHandler());
5656

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

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,17 @@
11
package love.wangqi.codec;
22

3+
import io.netty.handler.codec.http.HttpMethod;
34
import io.netty.handler.codec.http.HttpRequest;
45
import io.netty.handler.codec.http.multipart.HttpPostRequestEncoder;
56
import love.wangqi.route.Route;
67

8+
import java.net.InetSocketAddress;
9+
import java.net.MalformedURLException;
710
import java.net.URL;
11+
import java.util.Arrays;
12+
13+
import static love.wangqi.context.Constants.HTTP;
14+
import static love.wangqi.context.Constants.HTTPS;
815

916
/**
1017
* @author: wangqi
@@ -23,4 +30,62 @@ public RequestHolder(Route route, URL url, HttpRequest request, HttpPostRequestE
2330
this.request = request;
2431
this.bodyRequestEncoder = bodyRequestEncoder;
2532
}
33+
34+
public String getHost() {
35+
if (url.getHost() == null) {
36+
throw new RuntimeException("no host found");
37+
}
38+
return url.getHost();
39+
}
40+
41+
public int getPort() {
42+
String protocol = url.getProtocol() == null ? HTTP : url.getProtocol();
43+
int port = url.getPort();
44+
if (port == -1) {
45+
if (HTTP.equalsIgnoreCase(protocol)) {
46+
port = 80;
47+
} else if (HTTPS.equalsIgnoreCase(protocol)) {
48+
port = 443;
49+
}
50+
}
51+
return port;
52+
}
53+
54+
public String getProtocol() {
55+
return url.getProtocol();
56+
}
57+
58+
public InetSocketAddress getSocketAddress() {
59+
return new InetSocketAddress(getHost(), getPort());
60+
}
61+
62+
@Override
63+
public int hashCode() {
64+
return Arrays.hashCode(new Object[]{route, url});
65+
}
66+
67+
@Override
68+
public boolean equals(Object obj) {
69+
if (this == obj) {
70+
return true;
71+
}
72+
if (obj instanceof RequestHolder) {
73+
return this.route.equals(((RequestHolder) obj).route) && this.url.equals(((RequestHolder) obj).url);
74+
}
75+
return false;
76+
}
77+
78+
public static void main(String[] args) throws MalformedURLException {
79+
RequestHolder requestHolder1 = new RequestHolder(new Route(9L, HttpMethod.GET, "/html", new URL("http://10.100.64.71/html/user.json")), new URL("http://10.100.64.71/html/user.json"), null, null);
80+
RequestHolder requestHolder2 = new RequestHolder(new Route(9L, HttpMethod.GET, "/html", new URL("http://10.100.64.71/html/user.json")), new URL("http://10.100.64.71/html/user.json"), null, null);
81+
System.out.println(requestHolder1.hashCode());
82+
System.out.println(requestHolder2.hashCode());
83+
System.out.println(requestHolder1.equals(requestHolder2));
84+
85+
InetSocketAddress socketAddress1 = new InetSocketAddress("127.0.0.1", 80);
86+
InetSocketAddress socketAddress2 = new InetSocketAddress("127.0.0.1", 80);
87+
System.out.println(socketAddress1.hashCode());
88+
System.out.println(socketAddress2.hashCode());
89+
System.out.println(socketAddress1.equals(socketAddress2));
90+
}
2691
}

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

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,14 @@
11
package love.wangqi.config;
22

33

4+
import io.netty.channel.ChannelFutureListener;
45
import io.netty.channel.ChannelInboundHandler;
56
import io.netty.channel.ChannelOutboundHandler;
67
import love.wangqi.codec.HttpRequestBuilder;
78
import love.wangqi.core.ResponseHandler;
89
import love.wangqi.exception.handler.ExceptionHandler;
910
import love.wangqi.filter.FilterRegistry;
1011
import love.wangqi.filter.GatewayFilter;
11-
import love.wangqi.listener.ChannelCloseFutureListener;
1212
import love.wangqi.route.RouteMapper;
1313

1414
import java.util.*;
@@ -21,7 +21,7 @@
2121
*/
2222
public class GatewayConfig {
2323

24-
final static GatewayConfig INSTANCE = new GatewayConfig();
24+
public final static GatewayConfig INSTANCE = new GatewayConfig();
2525

2626
private final ConcurrentHashMap<String, List<GatewayFilter>> hashFiltersByType = new ConcurrentHashMap<String, List<GatewayFilter>>();
2727

@@ -74,9 +74,9 @@ public List<GatewayFilter> getFiltersByType(String filterType) {
7474
*/
7575
private int port;
7676
/**
77-
* channel关闭监听器
77+
* channel写完成监听器
7878
*/
79-
private ChannelCloseFutureListener channelCloseFutureListener;
79+
private ChannelFutureListener channelWriteFinishListener;
8080
/**
8181
* 异常处理器
8282
*/
@@ -141,12 +141,12 @@ public void setPort(int port) {
141141
this.port = port;
142142
}
143143

144-
public ChannelCloseFutureListener getChannelCloseFutureListener() {
145-
return channelCloseFutureListener;
144+
public ChannelFutureListener getChannelWriteFinishListener() {
145+
return channelWriteFinishListener;
146146
}
147147

148-
public void setChannelCloseFutureListener(ChannelCloseFutureListener channelCloseFutureListener) {
149-
this.channelCloseFutureListener = channelCloseFutureListener;
148+
public void setChannelWriteFinishListener(ChannelFutureListener channelWriteFinishListener) {
149+
this.channelWriteFinishListener = channelWriteFinishListener;
150150
}
151151

152152
public ExceptionHandler getExceptionHandler() {

‎src/main/java/love/wangqi/context/Attributes.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package love.wangqi.context;
22

3+
import io.netty.channel.Channel;
34
import io.netty.handler.codec.http.FullHttpRequest;
45
import io.netty.handler.codec.http.FullHttpResponse;
56
import io.netty.util.AttributeKey;
@@ -14,4 +15,5 @@ public interface Attributes {
1415
AttributeKey<Exception> EXCEPTION = AttributeKey.newInstance("exception");
1516
AttributeKey<FullHttpResponse> RESPONSE = AttributeKey.newInstance("response");
1617
AttributeKey<Boolean> KEEPALIVE = AttributeKey.newInstance("keepAlive");
18+
AttributeKey<Channel> SERVER_CHANNEL = AttributeKey.newInstance("serverChannel");
1719
}

‎src/main/java/love/wangqi/context/Constants.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
package love.wangqi.context;
2+
3+
/**
4+
* @author: wangqi
5+
* @description:
6+
* @date: Created in 2018-11-27 16:48
7+
*/
8+
public class Constants {
9+
public static final String HTTP = "http";
10+
public static final String HTTPS = "https";
11+
}

‎src/main/java/love/wangqi/core/DefaultChannelCloseFutureListener.java‎

Lines changed: 0 additions & 25 deletions
This file was deleted.

‎src/main/java/love/wangqi/core/DefaultChannelWriteFinishListener.java‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
package love.wangqi.core;
2+
3+
import io.netty.channel.Channel;
4+
import io.netty.channel.ChannelFuture;
5+
import io.netty.channel.ChannelFutureListener;
6+
import love.wangqi.context.ContextUtil;
7+
import org.slf4j.Logger;
8+
import org.slf4j.LoggerFactory;
9+
10+
/**
11+
* @author: wangqi
12+
* @description:
13+
* @date: Created in 2018-11-27 18:39
14+
*/
15+
public class DefaultChannelWriteFinishListener implements ChannelFutureListener {
16+
private Logger logger = LoggerFactory.getLogger(DefaultChannelWriteFinishListener.class);
17+
18+
@Override
19+
public void operationComplete(ChannelFuture future) throws Exception {
20+
Channel channel = future.channel();
21+
Boolean keepAlive = ContextUtil.getKeepAlive(channel);
22+
logger.debug("======= serverChannelId: {}", channel.id());
23+
24+
ContextUtil.getRequest(channel).release();
25+
26+
if (!keepAlive) {
27+
channel.close();
28+
}
29+
}
30+
}

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

Lines changed: 6 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22

33
import io.netty.channel.Channel;
44
import io.netty.channel.ChannelFuture;
5-
import io.netty.channel.ChannelFutureListener;
65
import io.netty.handler.codec.http.FullHttpResponse;
76
import io.netty.handler.codec.http.HttpHeaderNames;
87
import io.netty.handler.codec.http.HttpUtil;
@@ -22,16 +21,14 @@ public class ResponseHandler {
2221
private GatewayConfig config = GatewayConfig.getInstance();
2322

2423
public void send(Channel channel, FullHttpResponse response) {
24+
logger.debug("======= serverChannelId: {}", channel.id());
25+
logger.debug("readableBytes {}", response.content().readableBytes());
2526
response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
2627
Boolean keepAlive = ContextUtil.getKeepAlive(channel);
2728
HttpUtil.setKeepAlive(response, keepAlive == null ? false : keepAlive);
28-
channel.writeAndFlush(response).addListener(new ChannelFutureListener() {
29-
@Override
30-
public void operationComplete(ChannelFuture future) throws Exception {
31-
if (config.getChannelCloseFutureListener() != null) {
32-
config.getChannelCloseFutureListener().operationComplete(channel, future);
33-
}
34-
}
35-
});
29+
ChannelFuture future = channel.writeAndFlush(response);
30+
if (config.getChannelWriteFinishListener() != null) {
31+
future.addListener(config.getChannelWriteFinishListener());
32+
}
3633
}
3734
}

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@
88
import org.slf4j.Logger;
99
import org.slf4j.LoggerFactory;
1010

11+
import java.nio.charset.Charset;
12+
1113
/**
1214
* @author: wangqi
1315
* @description:

‎src/main/java/love/wangqi/filter/command/ForwardRunner.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ public class ForwardRunner {
2828
private final static String HTTPS = "https";
2929
private final static EventLoopGroup eventExecutors = new NioEventLoopGroup(8 * 8);
3030

31+
3132
private final static Logger logger = LoggerFactory.getLogger(ForwardRunner.class);
3233

3334
public ForwardRunner(Channel serverChannel, RequestHolder requestHolder) {
@@ -73,7 +74,7 @@ private UrlMetadata getProtocol(URL url) {
7374
return new UrlMetadata(protocol, host, port);
7475
}
7576

76-
private void forward() throws Exception {
77+
private void forward2() throws Exception {
7778
URL url = requestHolder.url;
7879
HttpRequest request = requestHolder.request;
7980
HttpPostRequestEncoder bodyRequestEncoder = requestHolder.bodyRequestEncoder;
@@ -107,4 +108,8 @@ public void operationComplete(ChannelFuture future) throws Exception {
107108
}
108109
});
109110
}
111+
112+
private void forward() throws Exception {
113+
HttpClientPool.INSTANCE.request(requestHolder, serverChannel);
114+
}
110115
}

0 commit comments

Comments
 (0)