Skip to content

Commit 083dece

Browse files
rstoyanchevrstoyanchev
committed
Polish netty client support
Mostly updates based on: https://github.com/spring-projects/spring-framework/wiki/Spring-Framework-Code-Style
1 parent e120757 commit 083dece

7 files changed

+85
-83
lines changed

‎spring-web/src/main/java/org/springframework/http/client/Netty4ClientHttpRequest.java

+34-37
Original file line numberDiff line numberDiff line change
@@ -60,13 +60,15 @@ class Netty4ClientHttpRequest extends AbstractAsyncClientHttpRequest implements
6060

6161
private final ByteBufOutputStream body;
6262

63+
6364
Netty4ClientHttpRequest(Bootstrap bootstrap, URI uri, HttpMethod method, int maxRequestSize) {
6465
this.bootstrap = bootstrap;
6566
this.uri = uri;
6667
this.method = method;
6768
this.body = new ByteBufOutputStream(Unpooled.buffer(maxRequestSize));
6869
}
6970

71+
7072
@Override
7173
public HttpMethod getMethod() {
7274
return this.method;
@@ -83,8 +85,7 @@ protected OutputStream getBodyInternal(HttpHeaders headers) throws IOException {
8385
}
8486

8587
@Override
86-
protected ListenableFuture<ClientHttpResponse> executeInternal(final HttpHeaders headers)
87-
throws IOException {
88+
protected ListenableFuture<ClientHttpResponse> executeInternal(final HttpHeaders headers) throws IOException {
8889
final SettableListenableFuture<ClientHttpResponse> responseFuture =
8990
new SettableListenableFuture<ClientHttpResponse>();
9091

@@ -93,42 +94,19 @@ protected ListenableFuture<ClientHttpResponse> executeInternal(final HttpHeaders
9394
public void operationComplete(ChannelFuture future) throws Exception {
9495
if (future.isSuccess()) {
9596
Channel channel = future.channel();
96-
channel.pipeline()
97-
.addLast(new SimpleChannelInboundHandler<FullHttpResponse>() {
98-
99-
@Override
100-
protected void channelRead0(
101-
ChannelHandlerContext ctx,
102-
FullHttpResponse msg) throws Exception {
103-
responseFuture
104-
.set(new Netty4ClientHttpResponse(ctx,
105-
msg));
106-
}
107-
108-
@Override
109-
public void exceptionCaught(
110-
ChannelHandlerContext ctx,
111-
Throwable cause) throws Exception {
112-
responseFuture.setException(cause);
113-
}
114-
});
115-
116-
FullHttpRequest nettyRequest =
117-
createFullHttpRequest(headers);
118-
97+
channel.pipeline().addLast(new RequestExecuteHandler(responseFuture));
98+
FullHttpRequest nettyRequest = createFullHttpRequest(headers);
11999
channel.writeAndFlush(nettyRequest);
120100
}
121101
else {
122102
responseFuture.setException(future.cause());
123103
}
124-
125104
}
126105
};
127106

128-
bootstrap.connect(uri.getHost(), getPort(uri)).addListener(connectionListener);
107+
this.bootstrap.connect(this.uri.getHost(), getPort(this.uri)).addListener(connectionListener);
129108

130109
return responseFuture;
131-
132110
}
133111

134112
@Override
@@ -142,7 +120,8 @@ public ClientHttpResponse execute() throws IOException {
142120
catch (ExecutionException ex) {
143121
if (ex.getCause() instanceof IOException) {
144122
throw (IOException) ex.getCause();
145-
} else {
123+
}
124+
else {
146125
throw new IOException(ex.getMessage(), ex);
147126
}
148127
}
@@ -163,17 +142,13 @@ else if ("https".equalsIgnoreCase(uri.getScheme())) {
163142

164143
private FullHttpRequest createFullHttpRequest(HttpHeaders headers) {
165144
io.netty.handler.codec.http.HttpMethod nettyMethod =
166-
io.netty.handler.codec.http.HttpMethod.valueOf(method.name());
145+
io.netty.handler.codec.http.HttpMethod.valueOf(this.method.name());
167146

168147
FullHttpRequest nettyRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1,
169-
nettyMethod, this.uri.getRawPath(),
170-
this.body.buffer());
148+
nettyMethod, this.uri.getRawPath(), this.body.buffer());
171149

172-
nettyRequest.headers()
173-
.set(io.netty.handler.codec.http.HttpHeaders.Names.HOST, uri.getHost());
174-
nettyRequest.headers()
175-
.set(io.netty.handler.codec.http.HttpHeaders.Names.CONNECTION,
176-
io.netty.handler.codec.http.HttpHeaders.Values.CLOSE);
150+
nettyRequest.headers().set(HttpHeaders.HOST, uri.getHost());
151+
nettyRequest.headers().set(HttpHeaders.CONNECTION, io.netty.handler.codec.http.HttpHeaders.Values.CLOSE);
177152

178153
for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
179154
nettyRequest.headers().add(entry.getKey(), entry.getValue());
@@ -183,4 +158,26 @@ private FullHttpRequest createFullHttpRequest(HttpHeaders headers) {
183158
}
184159

185160

161+
/**
162+
* A SimpleChannelInboundHandler to update the given SettableListenableFuture.
163+
*/
164+
private static class RequestExecuteHandler extends SimpleChannelInboundHandler<FullHttpResponse> {
165+
166+
private final SettableListenableFuture<ClientHttpResponse> responseFuture;
167+
168+
public RequestExecuteHandler(SettableListenableFuture<ClientHttpResponse> responseFuture) {
169+
this.responseFuture = responseFuture;
170+
}
171+
172+
@Override
173+
protected void channelRead0(ChannelHandlerContext context, FullHttpResponse response) throws Exception {
174+
this.responseFuture.set(new Netty4ClientHttpResponse(context, response));
175+
}
176+
177+
@Override
178+
public void exceptionCaught(ChannelHandlerContext context, Throwable cause) throws Exception {
179+
this.responseFuture.setException(cause);
180+
}
181+
}
182+
186183
}

‎spring-web/src/main/java/org/springframework/http/client/Netty4ClientHttpRequestFactory.java

+26-25
Original file line numberDiff line numberDiff line change
@@ -45,16 +45,16 @@
4545
* @author Arjen Poutsma
4646
* @since 4.2
4747
*/
48-
public class Netty4ClientHttpRequestFactory
49-
implements ClientHttpRequestFactory, AsyncClientHttpRequestFactory,
50-
InitializingBean, DisposableBean {
48+
public class Netty4ClientHttpRequestFactory implements ClientHttpRequestFactory,
49+
AsyncClientHttpRequestFactory, InitializingBean, DisposableBean {
5150

5251
/**
5352
* The default maximum request size.
5453
* @see #setMaxRequestSize(int)
5554
*/
5655
public static final int DEFAULT_MAX_REQUEST_SIZE = 1024 * 1024 * 10;
5756

57+
5858
private final EventLoopGroup eventLoopGroup;
5959

6060
private final boolean defaultEventLoopGroup;
@@ -65,18 +65,19 @@ public class Netty4ClientHttpRequestFactory
6565

6666
private Bootstrap bootstrap;
6767

68+
6869
/**
69-
* Creates a new {@code Netty4ClientHttpRequestFactory} with a default
70+
* Create a new {@code Netty4ClientHttpRequestFactory} with a default
7071
* {@link NioEventLoopGroup}.
7172
*/
7273
public Netty4ClientHttpRequestFactory() {
7374
int ioWorkerCount = Runtime.getRuntime().availableProcessors() * 2;
74-
eventLoopGroup = new NioEventLoopGroup(ioWorkerCount);
75-
defaultEventLoopGroup = true;
75+
this.eventLoopGroup = new NioEventLoopGroup(ioWorkerCount);
76+
this.defaultEventLoopGroup = true;
7677
}
7778

7879
/**
79-
* Creates a new {@code Netty4ClientHttpRequestFactory} with the given
80+
* Create a new {@code Netty4ClientHttpRequestFactory} with the given
8081
* {@link EventLoopGroup}.
8182
*
8283
* <p><b>NOTE:</b> the given group will <strong>not</strong> be
@@ -89,17 +90,20 @@ public Netty4ClientHttpRequestFactory(EventLoopGroup eventLoopGroup) {
8990
this.defaultEventLoopGroup = false;
9091
}
9192

93+
9294
/**
93-
* Sets the default maximum request size. The default is
94-
* {@link #DEFAULT_MAX_REQUEST_SIZE}.
95+
* Set the default maximum request size.
96+
* <p>By default this is set to {@link #DEFAULT_MAX_REQUEST_SIZE}.
9597
* @see HttpObjectAggregator#HttpObjectAggregator(int)
9698
*/
9799
public void setMaxRequestSize(int maxRequestSize) {
98100
this.maxRequestSize = maxRequestSize;
99101
}
100102

101103
/**
102-
* Sets the SSL context.
104+
* Set the SSL context. When configured it is used to create and insert an
105+
* {@link io.netty.handler.ssl.SslHandler} in the channel pipeline.
106+
* <p>By default this is not set.
103107
*/
104108
public void setSslContext(SslContext sslContext) {
105109
this.sslContext = sslContext;
@@ -108,14 +112,14 @@ public void setSslContext(SslContext sslContext) {
108112
private Bootstrap getBootstrap() {
109113
if (this.bootstrap == null) {
110114
Bootstrap bootstrap = new Bootstrap();
111-
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
115+
bootstrap.group(this.eventLoopGroup).channel(NioSocketChannel.class)
112116
.handler(new ChannelInitializer<SocketChannel>() {
113117
@Override
114-
protected void initChannel(SocketChannel ch) throws Exception {
115-
ChannelPipeline pipeline = ch.pipeline();
118+
protected void initChannel(SocketChannel channel) throws Exception {
119+
ChannelPipeline pipeline = channel.pipeline();
116120

117121
if (sslContext != null) {
118-
pipeline.addLast(sslContext.newHandler(ch.alloc()));
122+
pipeline.addLast(sslContext.newHandler(channel.alloc()));
119123
}
120124
pipeline.addLast(new HttpClientCodec());
121125
pipeline.addLast(new HttpObjectAggregator(maxRequestSize));
@@ -131,29 +135,26 @@ public void afterPropertiesSet() throws Exception {
131135
getBootstrap();
132136
}
133137

134-
private Netty4ClientHttpRequest createRequestInternal(URI uri, HttpMethod httpMethod) {
135-
return new Netty4ClientHttpRequest(getBootstrap(), uri, httpMethod, maxRequestSize);
136-
}
137-
138138
@Override
139-
public ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod)
140-
throws IOException {
139+
public ClientHttpRequest createRequest(URI uri, HttpMethod httpMethod) throws IOException {
141140
return createRequestInternal(uri, httpMethod);
142141
}
143142

144143
@Override
145-
public AsyncClientHttpRequest createAsyncRequest(URI uri, HttpMethod httpMethod)
146-
throws IOException {
144+
public AsyncClientHttpRequest createAsyncRequest(URI uri, HttpMethod httpMethod) throws IOException {
147145
return createRequestInternal(uri, httpMethod);
148146
}
149147

148+
private Netty4ClientHttpRequest createRequestInternal(URI uri, HttpMethod httpMethod) {
149+
return new Netty4ClientHttpRequest(getBootstrap(), uri, httpMethod, this.maxRequestSize);
150+
}
151+
150152
@Override
151153
public void destroy() throws InterruptedException {
152-
if (defaultEventLoopGroup) {
154+
if (this.defaultEventLoopGroup) {
153155
// clean up the EventLoopGroup if we created it in the constructor
154-
eventLoopGroup.shutdownGracefully().sync();
156+
this.eventLoopGroup.shutdownGracefully().sync();
155157
}
156158
}
157159

158-
159160
}

‎spring-web/src/main/java/org/springframework/http/client/Netty4ClientHttpResponse.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -45,8 +45,7 @@ class Netty4ClientHttpResponse extends AbstractClientHttpResponse {
4545
private HttpHeaders headers;
4646

4747

48-
Netty4ClientHttpResponse(ChannelHandlerContext context,
49-
FullHttpResponse nettyResponse) {
48+
Netty4ClientHttpResponse(ChannelHandlerContext context, FullHttpResponse nettyResponse) {
5049
Assert.notNull(context, "'context' must not be null");
5150
Assert.notNull(nettyResponse, "'nettyResponse' must not be null");
5251
this.context = context;
@@ -55,6 +54,7 @@ class Netty4ClientHttpResponse extends AbstractClientHttpResponse {
5554
this.nettyResponse.retain();
5655
}
5756

57+
5858
@Override
5959
public int getRawStatusCode() throws IOException {
6060
return this.nettyResponse.getStatus().code();

‎spring-web/src/test/java/org/springframework/http/client/AbstractAsyncHttpRequestFactoryTestCase.java

+17-14
Original file line numberDiff line numberDiff line change
@@ -16,17 +16,19 @@
1616

1717
package org.springframework.http.client;
1818

19+
import static org.junit.Assert.assertEquals;
20+
import static org.junit.Assert.assertTrue;
21+
import static org.junit.Assert.fail;
22+
1923
import java.io.IOException;
2024
import java.net.URI;
2125
import java.util.Arrays;
2226
import java.util.Locale;
2327
import java.util.concurrent.Future;
2428

2529
import org.junit.After;
26-
import static org.junit.Assert.*;
2730
import org.junit.Before;
2831
import org.junit.Test;
29-
3032
import org.springframework.beans.factory.DisposableBean;
3133
import org.springframework.beans.factory.InitializingBean;
3234
import org.springframework.http.HttpMethod;
@@ -44,16 +46,16 @@ public abstract class AbstractAsyncHttpRequestFactoryTestCase extends AbstractJe
4446

4547
@Before
4648
public final void createFactory() throws Exception {
47-
factory = createRequestFactory();
48-
if (factory instanceof InitializingBean) {
49-
((InitializingBean) factory).afterPropertiesSet();
49+
this.factory = createRequestFactory();
50+
if (this.factory instanceof InitializingBean) {
51+
((InitializingBean) this.factory).afterPropertiesSet();
5052
}
5153
}
5254

5355
@After
5456
public final void destroyFactory() throws Exception {
55-
if (factory instanceof DisposableBean) {
56-
((DisposableBean) factory).destroy();
57+
if (this.factory instanceof DisposableBean) {
58+
((DisposableBean) this.factory).destroy();
5759
}
5860
}
5961

@@ -63,22 +65,23 @@ public final void destroyFactory() throws Exception {
6365
@Test
6466
public void status() throws Exception {
6567
URI uri = new URI(baseUrl + "/status/notfound");
66-
AsyncClientHttpRequest request = factory.createAsyncRequest(uri, HttpMethod.GET);
68+
AsyncClientHttpRequest request = this.factory.createAsyncRequest(uri, HttpMethod.GET);
6769
assertEquals("Invalid HTTP method", HttpMethod.GET, request.getMethod());
6870
assertEquals("Invalid HTTP URI", uri, request.getURI());
6971
Future<ClientHttpResponse> futureResponse = request.executeAsync();
7072
ClientHttpResponse response = futureResponse.get();
7173
try {
7274
assertEquals("Invalid status code", HttpStatus.NOT_FOUND, response.getStatusCode());
73-
} finally {
75+
}
76+
finally {
7477
response.close();
7578
}
7679
}
7780

7881
@Test
7982
public void statusCallback() throws Exception {
8083
URI uri = new URI(baseUrl + "/status/notfound");
81-
AsyncClientHttpRequest request = factory.createAsyncRequest(uri, HttpMethod.GET);
84+
AsyncClientHttpRequest request = this.factory.createAsyncRequest(uri, HttpMethod.GET);
8285
assertEquals("Invalid HTTP method", HttpMethod.GET, request.getMethod());
8386
assertEquals("Invalid HTTP URI", uri, request.getURI());
8487
ListenableFuture<ClientHttpResponse> listenableFuture = request.executeAsync();
@@ -108,7 +111,7 @@ public void onFailure(Throwable ex) {
108111

109112
@Test
110113
public void echo() throws Exception {
111-
AsyncClientHttpRequest request = factory.createAsyncRequest(new URI(baseUrl + "/echo"), HttpMethod.PUT);
114+
AsyncClientHttpRequest request = this.factory.createAsyncRequest(new URI(baseUrl + "/echo"), HttpMethod.PUT);
112115
assertEquals("Invalid HTTP method", HttpMethod.PUT, request.getMethod());
113116
String headerName = "MyHeader";
114117
String headerValue1 = "value1";
@@ -143,7 +146,7 @@ public void echo() throws Exception {
143146

144147
@Test
145148
public void multipleWrites() throws Exception {
146-
AsyncClientHttpRequest request = factory.createAsyncRequest(new URI(baseUrl + "/echo"), HttpMethod.POST);
149+
AsyncClientHttpRequest request = this.factory.createAsyncRequest(new URI(baseUrl + "/echo"), HttpMethod.POST);
147150
final byte[] body = "Hello World".getBytes("UTF-8");
148151

149152
if (request instanceof StreamingHttpOutputMessage) {
@@ -170,7 +173,7 @@ public void multipleWrites() throws Exception {
170173

171174
@Test
172175
public void headersAfterExecute() throws Exception {
173-
AsyncClientHttpRequest request = factory.createAsyncRequest(new URI(baseUrl + "/echo"), HttpMethod.POST);
176+
AsyncClientHttpRequest request = this.factory.createAsyncRequest(new URI(baseUrl + "/echo"), HttpMethod.POST);
174177
request.getHeaders().add("MyHeader", "value");
175178
byte[] body = "Hello World".getBytes("UTF-8");
176179
FileCopyUtils.copy(body, request.getBody());
@@ -202,7 +205,7 @@ public void httpMethods() throws Exception {
202205
protected void assertHttpMethod(String path, HttpMethod method) throws Exception {
203206
ClientHttpResponse response = null;
204207
try {
205-
AsyncClientHttpRequest request = factory.createAsyncRequest(new URI(baseUrl + "/methods/" + path), method);
208+
AsyncClientHttpRequest request = this.factory.createAsyncRequest(new URI(baseUrl + "/methods/" + path), method);
206209
Future<ClientHttpResponse> futureResponse = request.executeAsync();
207210
response = futureResponse.get();
208211
assertEquals("Invalid response status", HttpStatus.OK, response.getStatusCode());

‎spring-web/src/test/java/org/springframework/http/client/AbstractHttpRequestFactoryTestCase.java

+2-1
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,8 @@ public void status() throws Exception {
6969
ClientHttpResponse response = request.execute();
7070
try {
7171
assertEquals("Invalid status code", HttpStatus.NOT_FOUND, response.getStatusCode());
72-
} finally {
72+
}
73+
finally {
7374
response.close();
7475
}
7576
}

0 commit comments

Comments
 (0)