Skip to content

Commit ab36119

Browse files
author
Stephane Landelle
committed
Refactor NettyAsyncHttpProviderConfig, make non Netty properties explicit members, close AsyncHttpClient#164 
1 parent d482eca commit ab36119

File tree

5 files changed

+94
-69
lines changed

5 files changed

+94
-69
lines changed

‎providers/netty/src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProvider.java

Lines changed: 19 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -193,23 +193,20 @@ public NettyAsyncHttpProvider(AsyncHttpClientConfig config) {
193193
asyncHttpProviderConfig = new NettyAsyncHttpProviderConfig();
194194
}
195195

196-
if (asyncHttpProviderConfig.getProperty(NettyAsyncHttpProviderConfig.USE_BLOCKING_IO) != null) {
196+
if (asyncHttpProviderConfig.isUseBlockingIO()) {
197197
socketChannelFactory = new OioClientSocketChannelFactory(config.executorService());
198198
this.allowReleaseSocketChannelFactory = true;
199199
} else {
200200
// check if external NioClientSocketChannelFactory is defined
201-
Object oo = asyncHttpProviderConfig.getProperty(NettyAsyncHttpProviderConfig.SOCKET_CHANNEL_FACTORY);
202-
if (oo != null && NioClientSocketChannelFactory.class.isAssignableFrom(oo.getClass())) {
203-
this.socketChannelFactory = NioClientSocketChannelFactory.class.cast(oo);
201+
NioClientSocketChannelFactory scf = asyncHttpProviderConfig.getSocketChannelFactory();
202+
if (scf != null) {
203+
this.socketChannelFactory = scf;
204204

205205
// cannot allow releasing shared channel factory
206206
this.allowReleaseSocketChannelFactory = false;
207207
} else {
208-
ExecutorService e = null;
209-
Object o = asyncHttpProviderConfig.getProperty(NettyAsyncHttpProviderConfig.BOSS_EXECUTOR_SERVICE);
210-
if (o != null && ExecutorService.class.isAssignableFrom(o.getClass())) {
211-
e = ExecutorService.class.cast(o);
212-
} else {
208+
ExecutorService e = asyncHttpProviderConfig.getBossExecutorService();
209+
if (e == null) {
213210
e = Executors.newCachedThreadPool();
214211
}
215212
int numWorkers = config.getIoThreadMultiplier() * Runtime.getRuntime().availableProcessors();
@@ -256,18 +253,21 @@ public String toString() {
256253
void configureNetty() {
257254
if (asyncHttpProviderConfig != null) {
258255
for (Entry<String, Object> entry : asyncHttpProviderConfig.propertiesSet()) {
259-
plainBootstrap.setOption(entry.getKey(), entry.getValue());
256+
String key = entry.getKey();
257+
Object value = entry.getValue();
258+
plainBootstrap.setOption(key, value);
259+
webSocketBootstrap.setOption(key, value);
260+
secureBootstrap.setOption(key, value);
261+
secureWebSocketBootstrap.setOption(key, value);
260262
}
261263
}
262264

263265
plainBootstrap.setPipelineFactory(createPlainPipelineFactory());
264266
DefaultChannelFuture.setUseDeadLockChecker(false);
265267

266268
if (asyncHttpProviderConfig != null) {
267-
Object value = asyncHttpProviderConfig.getProperty(NettyAsyncHttpProviderConfig.EXECUTE_ASYNC_CONNECT);
268-
if (value != null && Boolean.class.isAssignableFrom(value.getClass())) {
269-
executeConnectAsync = Boolean.class.cast(value);
270-
} else if (asyncHttpProviderConfig.getProperty(NettyAsyncHttpProviderConfig.DISABLE_NESTED_REQUEST) != null) {
269+
executeConnectAsync = asyncHttpProviderConfig.isAsyncConnect();
270+
if (!executeConnectAsync) {
271271
DefaultChannelFuture.setUseDeadLockChecker(true);
272272
}
273273
}
@@ -352,13 +352,6 @@ public ChannelPipeline getPipeline() throws Exception {
352352
return pipeline;
353353
}
354354
});
355-
356-
if (asyncHttpProviderConfig != null) {
357-
for (Entry<String, Object> entry : asyncHttpProviderConfig.propertiesSet()) {
358-
secureBootstrap.setOption(entry.getKey(), entry.getValue());
359-
secureWebSocketBootstrap.setOption(entry.getKey(), entry.getValue());
360-
}
361-
}
362355
}
363356

364357
private Channel lookupInCache(URI uri, ConnectionPoolKeyStrategy connectionPoolKeyStrategy) {
@@ -1012,11 +1005,6 @@ private <T> ListenableFuture<T> doConnect(final Request request, final AsyncHand
10121005
ClientBootstrap bootstrap = request.getUrl().startsWith(WEBSOCKET) ? (useSSl ? secureWebSocketBootstrap : webSocketBootstrap) : (useSSl ? secureBootstrap : plainBootstrap);
10131006
bootstrap.setOption("connectTimeoutMillis", config.getConnectionTimeoutInMs());
10141007

1015-
// Do no enable this with win.
1016-
if (System.getProperty("os.name").toLowerCase().indexOf("win") == -1) {
1017-
bootstrap.setOption("reuseAddress", asyncHttpProviderConfig.getProperty(NettyAsyncHttpProviderConfig.REUSE_ADDRESS));
1018-
}
1019-
10201008
try {
10211009
InetSocketAddress remoteAddress;
10221010
if (request.getInetAddress() != null) {
@@ -1694,10 +1682,10 @@ public static <T> NettyResponseFuture<T> newFuture(URI uri,
16941682
private class ProgressListener implements ChannelFutureProgressListener {
16951683

16961684
private final boolean notifyHeaders;
1697-
private final AsyncHandler asyncHandler;
1685+
private final AsyncHandler<?> asyncHandler;
16981686
private final NettyResponseFuture<?> future;
16991687

1700-
public ProgressListener(boolean notifyHeaders, AsyncHandler asyncHandler, NettyResponseFuture<?> future) {
1688+
public ProgressListener(boolean notifyHeaders, AsyncHandler<?> asyncHandler, NettyResponseFuture<?> future) {
17011689
this.notifyHeaders = notifyHeaders;
17021690
this.asyncHandler = asyncHandler;
17031691
this.future = future;
@@ -2166,14 +2154,7 @@ public void handle(final ChannelHandlerContext ctx, final MessageEvent e) throws
21662154
newRealm = kerberosChallenge(wwwAuth, request, proxyServer, headers, realm, future);
21672155
if (newRealm == null) return;
21682156
} else {
2169-
Realm.RealmBuilder realmBuilder;
2170-
if (realm != null) {
2171-
realmBuilder = new Realm.RealmBuilder().clone(realm).setScheme(realm.getAuthScheme())
2172-
;
2173-
} else {
2174-
realmBuilder = new Realm.RealmBuilder();
2175-
}
2176-
newRealm = realmBuilder
2157+
newRealm = new Realm.RealmBuilder().clone(realm).setScheme(realm.getAuthScheme())
21772158
.setUri(request.getURI().getPath())
21782159
.setMethodName(request.getMethod())
21792160
.setUsePreemptiveAuth(true)
@@ -2325,18 +2306,15 @@ public void onClose(ChannelHandlerContext ctx, ChannelStateEvent e) {
23252306
}
23262307

23272308
private final class WebSocketProtocol implements Protocol {
2328-
private static final byte OPCODE_CONT = 0x0;
23292309
private static final byte OPCODE_TEXT = 0x1;
23302310
private static final byte OPCODE_BINARY = 0x2;
23312311
private static final byte OPCODE_UNKNOWN = -1;
23322312

2333-
protected ChannelBuffer byteBuffer = null;
2334-
protected StringBuilder textBuffer = null;
23352313
protected byte pendingOpcode = OPCODE_UNKNOWN;
23362314

23372315
// @Override
23382316
public void handle(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
2339-
NettyResponseFuture future = NettyResponseFuture.class.cast(ctx.getAttachment());
2317+
NettyResponseFuture<?> future = NettyResponseFuture.class.cast(ctx.getAttachment());
23402318
WebSocketUpgradeHandler h = WebSocketUpgradeHandler.class.cast(future.getAsyncHandler());
23412319
Request request = future.getRequest();
23422320

@@ -2345,7 +2323,7 @@ public void handle(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
23452323

23462324
HttpResponseStatus s = new ResponseStatus(future.getURI(), response, NettyAsyncHttpProvider.this);
23472325
HttpResponseHeaders responseHeaders = new ResponseHeaders(future.getURI(), response, NettyAsyncHttpProvider.this);
2348-
FilterContext<?> fc = new FilterContext.FilterContextBuilder()
2326+
FilterContext fc = new FilterContext.FilterContextBuilder()
23492327
.asyncHandler(h)
23502328
.request(request)
23512329
.responseStatus(s)

‎providers/netty/src/main/java/com/ning/http/client/providers/netty/NettyAsyncHttpProviderConfig.java

Lines changed: 71 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -16,73 +16,88 @@
1616
*/
1717
package com.ning.http.client.providers.netty;
1818

19-
import com.ning.http.client.AsyncHttpProviderConfig;
20-
19+
import java.util.HashMap;
2120
import java.util.Map;
2221
import java.util.Set;
23-
import java.util.concurrent.ConcurrentHashMap;
22+
import java.util.concurrent.ExecutorService;
23+
24+
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
25+
import org.slf4j.Logger;
26+
import org.slf4j.LoggerFactory;
27+
28+
import com.ning.http.client.AsyncHttpProviderConfig;
2429

2530
/**
2631
* This class can be used to pass Netty's internal configuration options. See Netty documentation for more information.
2732
*/
2833
public class NettyAsyncHttpProviderConfig implements AsyncHttpProviderConfig<String, Object> {
2934

35+
private final static Logger LOGGER = LoggerFactory.getLogger(NettyAsyncHttpProviderConfig.class);
36+
3037
/**
3138
* Use Netty's blocking IO stategy.
3239
*/
33-
public final static String USE_BLOCKING_IO = "useBlockingIO";
40+
private boolean useBlockingIO;
3441

3542
/**
36-
* Use direct {@link java.nio.ByteBuffer}
43+
* Allow configuring the Netty's socket channel factory.
3744
*/
38-
public final static String USE_DIRECT_BYTEBUFFER = "bufferFactory";
45+
private NioClientSocketChannelFactory socketChannelFactory;
3946

4047
/**
41-
* Execute the connect operation asynchronously.
48+
* Allow configuring the Netty's boss executor service.
4249
*/
43-
public final static String EXECUTE_ASYNC_CONNECT = "asyncConnect";
50+
private ExecutorService bossExecutorService;
4451

4552
/**
46-
* Allow nested request from any {@link com.ning.http.client.AsyncHandler}
53+
* Execute the connect operation asynchronously.
4754
*/
48-
public final static String DISABLE_NESTED_REQUEST = "disableNestedRequest";
55+
private boolean asyncConnect;
4956

5057
/**
51-
* Allow configuring the Netty's boss executor service.
58+
* Use direct {@link java.nio.ByteBuffer}
5259
*/
53-
public final static String BOSS_EXECUTOR_SERVICE = "bossExecutorService";
54-
60+
public final static String USE_DIRECT_BYTEBUFFER = "bufferFactory";
61+
5562
/**
56-
* Allow configuring the Netty's socket channel factory.
63+
* Allow nested request from any {@link com.ning.http.client.AsyncHandler}
5764
*/
58-
public final static String SOCKET_CHANNEL_FACTORY = "socketChannelFactory";
65+
public final static String DISABLE_NESTED_REQUEST = "disableNestedRequest";
5966

6067
/**
6168
* See {@link java.net.Socket#setReuseAddress(boolean)}
6269
*/
6370
public final static String REUSE_ADDRESS = "reuseAddress";
6471

65-
private final ConcurrentHashMap<String, Object> properties = new ConcurrentHashMap<String, Object>();
72+
private final Map<String, Object> properties = new HashMap<String, Object>();
6673

6774
public NettyAsyncHttpProviderConfig() {
68-
properties.put(REUSE_ADDRESS, "false");
75+
properties.put(REUSE_ADDRESS, Boolean.FALSE);
6976
}
7077

7178
/**
7279
* Add a property that will be used when the AsyncHttpClient initialize its {@link com.ning.http.client.AsyncHttpProvider}
73-
*
74-
* @param name the name of the property
75-
* @param value the value of the property
80+
*
81+
* @param name
82+
* the name of the property
83+
* @param value
84+
* the value of the property
7685
* @return this instance of AsyncHttpProviderConfig
7786
*/
7887
public NettyAsyncHttpProviderConfig addProperty(String name, Object value) {
79-
properties.put(name, value);
88+
89+
if (name.equals(REUSE_ADDRESS) && value == Boolean.TRUE && System.getProperty("os.name").toLowerCase().contains("win")) {
90+
LOGGER.warn("Can't enable {} on Windows", REUSE_ADDRESS);
91+
} else {
92+
properties.put(name, value);
93+
}
94+
8095
return this;
8196
}
8297

8398
/**
8499
* Return the value associated with the property's name
85-
*
100+
*
86101
* @param name
87102
* @return this instance of AsyncHttpProviderConfig
88103
*/
@@ -92,7 +107,7 @@ public Object getProperty(String name) {
92107

93108
/**
94109
* Remove the value associated with the property's name
95-
*
110+
*
96111
* @param name
97112
* @return true if removed
98113
*/
@@ -102,10 +117,42 @@ public Object removeProperty(String name) {
102117

103118
/**
104119
* Return the curent entry set.
105-
*
120+
*
106121
* @return a the curent entry set.
107122
*/
108123
public Set<Map.Entry<String, Object>> propertiesSet() {
109124
return properties.entrySet();
110125
}
126+
127+
public boolean isUseBlockingIO() {
128+
return useBlockingIO;
129+
}
130+
131+
public void setUseBlockingIO(boolean useBlockingIO) {
132+
this.useBlockingIO = useBlockingIO;
133+
}
134+
135+
public NioClientSocketChannelFactory getSocketChannelFactory() {
136+
return socketChannelFactory;
137+
}
138+
139+
public void setSocketChannelFactory(NioClientSocketChannelFactory socketChannelFactory) {
140+
this.socketChannelFactory = socketChannelFactory;
141+
}
142+
143+
public ExecutorService getBossExecutorService() {
144+
return bossExecutorService;
145+
}
146+
147+
public void setBossExecutorService(ExecutorService bossExecutorService) {
148+
this.bossExecutorService = bossExecutorService;
149+
}
150+
151+
public boolean isAsyncConnect() {
152+
return asyncConnect;
153+
}
154+
155+
public void setAsyncConnect(boolean asyncConnect) {
156+
this.asyncConnect = asyncConnect;
157+
}
111158
}

‎providers/netty/src/test/java/com/ning/http/client/providers/netty/NettyAsyncHttpProviderTest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,7 @@ public class NettyAsyncHttpProviderTest extends AbstractBasicTest {
2828
@Test
2929
public void bossThreadPoolExecutor() throws Throwable {
3030
NettyAsyncHttpProviderConfig conf = new NettyAsyncHttpProviderConfig();
31-
conf.addProperty(NettyAsyncHttpProviderConfig.BOSS_EXECUTOR_SERVICE, Executors.newSingleThreadExecutor());
31+
conf.setBossExecutorService(Executors.newSingleThreadExecutor());
3232

3333
AsyncHttpClientConfig cf = new AsyncHttpClientConfig.Builder().setAsyncHttpClientProviderConfig(conf).build();
3434
AsyncHttpClient c = getAsyncHttpClient(cf);

‎providers/netty/src/test/java/com/ning/http/client/providers/netty/NettyRedirectConnectionUsageTest.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ public AsyncHttpClient getAsyncHttpClient(AsyncHttpClientConfig config) {
2727
protected AsyncHttpProviderConfig getProviderConfig() {
2828
final NettyAsyncHttpProviderConfig config = new NettyAsyncHttpProviderConfig();
2929
if (System.getProperty("blockingio") != null) {
30-
config.addProperty(NettyAsyncHttpProviderConfig.USE_BLOCKING_IO, "true");
30+
config.setUseBlockingIO(true);
3131
}
3232
return config;
3333
}

‎providers/netty/src/test/java/com/ning/http/client/providers/netty/RetryNonBlockingIssue.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -185,7 +185,7 @@ public void testRetryNonBlockingAsyncConnect() throws IOException, InterruptedEx
185185

186186
NettyAsyncHttpProviderConfig config = new
187187
NettyAsyncHttpProviderConfig();
188-
config.addProperty(NettyAsyncHttpProviderConfig.EXECUTE_ASYNC_CONNECT, "true");
188+
config.setAsyncConnect(true);
189189

190190
bc.setAsyncHttpClientProviderConfig(config);
191191
c = new AsyncHttpClient(bc.build());
@@ -230,7 +230,7 @@ public void testRetryBlocking() throws IOException, InterruptedException,
230230

231231
NettyAsyncHttpProviderConfig config = new
232232
NettyAsyncHttpProviderConfig();
233-
config.addProperty(NettyAsyncHttpProviderConfig.USE_BLOCKING_IO, "true");
233+
config.setUseBlockingIO(true);
234234

235235
bc.setAsyncHttpClientProviderConfig(config);
236236
c = new AsyncHttpClient(bc.build());

0 commit comments

Comments
 (0)