Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit4d02959

Browse files
author
Stephane Landelle
committed
Reimplement Netty provider timeouts,closeAsyncHttpClient#457
1 parent19a1fd5 commit4d02959

File tree

9 files changed

+276
-152
lines changed

9 files changed

+276
-152
lines changed

‎providers/netty/src/main/java/org/asynchttpclient/providers/netty/channel/Channels.java‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,9 @@
3737
importio.netty.handler.stream.ChunkedWriteHandler;
3838
importio.netty.util.Attribute;
3939
importio.netty.util.AttributeKey;
40+
importio.netty.util.HashedWheelTimer;
41+
importio.netty.util.Timeout;
42+
importio.netty.util.TimerTask;
4043

4144
importjava.io.IOException;
4245
importjava.lang.reflect.Field;
@@ -46,6 +49,7 @@
4649
importjava.util.Map;
4750
importjava.util.Map.Entry;
4851
importjava.util.concurrent.Semaphore;
52+
importjava.util.concurrent.TimeUnit;
4953

5054
importjavax.net.ssl.SSLEngine;
5155

@@ -104,6 +108,8 @@ public boolean remove(Object o) {
104108
}
105109
};
106110

111+
privatefinalHashedWheelTimerhashedWheelTimer;
112+
107113
publicChannels(finalAsyncHttpClientConfigconfig,NettyAsyncHttpProviderConfigasyncHttpProviderConfig) {
108114

109115
this.config =config;
@@ -176,6 +182,9 @@ public Channels(final AsyncHttpClientConfig config, NettyAsyncHttpProviderConfig
176182
webSocketBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,timeOut);
177183
secureBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,timeOut);
178184
secureWebSocketBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,timeOut);
185+
186+
hashedWheelTimer =newHashedWheelTimer();
187+
hashedWheelTimer.start();
179188
}
180189

181190
privateSSLEnginecreateSSLEngine()throwsIOException,GeneralSecurityException {
@@ -271,13 +280,15 @@ public void close() {
271280
Objectattribute =getDefaultAttribute(channel);
272281
if (attributeinstanceofNettyResponseFuture<?>) {
273282
NettyResponseFuture<?>future = (NettyResponseFuture<?>)attribute;
274-
future.setReaperFuture(null);
283+
future.cancelTimeouts();
275284
}
276285
}
277286
openChannels.close();
278287
if (allowReleaseEventLoopGroup) {
279288
eventLoopGroup.shutdownGracefully();
280289
}
290+
291+
hashedWheelTimer.stop();
281292
}
282293

283294
// some servers can use the same port for HTTP and HTTPS
@@ -456,6 +467,10 @@ public void abort(NettyResponseFuture<?> future, Throwable t) {
456467
future.abort(t);
457468
}
458469

470+
publicTimeoutnewTimeoutInMs(TimerTasktask,longdelayInMs) {
471+
returnhashedWheelTimer.newTimeout(task,delayInMs,TimeUnit.MILLISECONDS);
472+
}
473+
459474
publicstaticSslHandlergetSslHandler(Channelchannel) {
460475
return (SslHandler)channel.pipeline().get(Channels.SSL_HANDLER);
461476
}

‎providers/netty/src/main/java/org/asynchttpclient/providers/netty/future/FutureReaper.java‎

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

‎providers/netty/src/main/java/org/asynchttpclient/providers/netty/future/NettyResponseFuture.java‎

Lines changed: 34 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,7 @@
4141
importorg.asynchttpclient.providers.netty.DiscardEvent;
4242
importorg.asynchttpclient.providers.netty.channel.Channels;
4343
importorg.asynchttpclient.providers.netty.request.NettyRequest;
44+
importorg.asynchttpclient.providers.netty.request.timeout.TimeoutsHolder;
4445
importorg.slf4j.Logger;
4546
importorg.slf4j.LoggerFactory;
4647

@@ -59,7 +60,8 @@ public enum STATE {
5960
}
6061

6162
privatefinalintrequestTimeoutInMs;
62-
privatefinalAsyncHttpClientConfigconfig;
63+
privatevolatilebooleanrequestTimeoutReached;
64+
privatevolatilebooleanidleConnectionTimeoutReached;
6365
privatefinallongstart =millisTime();
6466
privatefinalConnectionPoolKeyStrategyconnectionPoolKeyStrategy;
6567
privatefinalProxyServerproxyServer;
@@ -80,7 +82,7 @@ public enum STATE {
8082
privatefinalAtomicBooleanthrowableCalled =newAtomicBoolean(false);
8183
privatefinalAtomicReference<V>content =newAtomicReference<V>();
8284
privatefinalAtomicReference<ExecutionException>exEx =newAtomicReference<ExecutionException>();
83-
privatevolatileFutureReaperreaperFuture;
85+
privatevolatileTimeoutsHoldertimeoutsHolder;
8486

8587
// state mutated only inside the event loop
8688
privateChannelchannel;
@@ -111,7 +113,6 @@ public NettyResponseFuture(URI uri,//
111113
this.request =request;
112114
this.nettyRequest =nettyRequest;
113115
this.uri =uri;
114-
this.config =config;
115116
this.connectionPoolKeyStrategy =connectionPoolKeyStrategy;
116117
this.proxyServer =proxyServer;
117118

@@ -156,7 +157,7 @@ public void setAsyncHandler(AsyncHandler<V> asyncHandler) {
156157

157158
@Override
158159
publicbooleancancel(booleanforce) {
159-
cancelReaper();
160+
cancelTimeouts();
160161

161162
if (isCancelled.get())
162163
returnfalse;
@@ -186,31 +187,39 @@ public boolean cancel(boolean force) {
186187
* @return <code>true</code> if response has expired and should be terminated.
187188
*/
188189
publicbooleanhasExpired() {
189-
longnow =millisTime();
190-
returnhasConnectionIdleTimedOut(now) ||hasRequestTimedOut(now);
190+
returnrequestTimeoutReached ||idleConnectionTimeoutReached;
191191
}
192192

193-
publicbooleanhasConnectionIdleTimedOut(longnow) {
194-
returnconfig.getIdleConnectionTimeoutInMs() != -1 && (now -touch.get()) >=config.getIdleConnectionTimeoutInMs();
193+
publicvoidsetRequestTimeoutReached() {
194+
this.requestTimeoutReached =true;
195195
}
196196

197-
publicbooleanhasRequestTimedOut(longnow) {
198-
returnrequestTimeoutInMs != -1 && (now -start) >=requestTimeoutInMs;
197+
publicbooleanisRequestTimeoutReached() {
198+
returnrequestTimeoutReached;
199+
}
200+
201+
publicvoidsetIdleConnectionTimeoutReached() {
202+
this.idleConnectionTimeoutReached =true;
203+
}
204+
205+
publicbooleanisIdleConnectionTimeoutReached() {
206+
returnidleConnectionTimeoutReached;
199207
}
200208

201209
@Override
202210
publicVget()throwsInterruptedException,ExecutionException {
203211
try {
204212
returnget(requestTimeoutInMs,TimeUnit.MILLISECONDS);
205213
}catch (TimeoutExceptione) {
206-
cancelReaper();
214+
cancelTimeouts();
207215
thrownewExecutionException(e);
208216
}
209217
}
210218

211-
publicvoidcancelReaper() {
212-
if (reaperFuture !=null) {
213-
reaperFuture.cancel(false);
219+
publicvoidcancelTimeouts() {
220+
if (timeoutsHolder !=null) {
221+
timeoutsHolder.cancel();
222+
timeoutsHolder =null;
214223
}
215224
}
216225

@@ -242,7 +251,7 @@ public V get(long l, TimeUnit tu) throws InterruptedException, TimeoutException,
242251
}
243252
thrownewExecutionException(te);
244253
}finally {
245-
cancelReaper();
254+
cancelTimeouts();
246255
}
247256
}
248257
}
@@ -278,7 +287,7 @@ private V getContent() throws ExecutionException {
278287
}
279288
thrownewRuntimeException(ex);
280289
}finally {
281-
cancelReaper();
290+
cancelTimeouts();
282291
}
283292
}
284293
}
@@ -290,7 +299,7 @@ private V getContent() throws ExecutionException {
290299
publicfinalvoiddone() {
291300

292301
try {
293-
cancelReaper();
302+
cancelTimeouts();
294303

295304
if (exEx.get() !=null) {
296305
return;
@@ -311,7 +320,7 @@ public final void done() {
311320
}
312321

313322
publicfinalvoidabort(finalThrowablet) {
314-
cancelReaper();
323+
cancelTimeouts();
315324

316325
if (isDone.get() ||isCancelled.get())
317326
return;
@@ -366,9 +375,8 @@ public int incrementAndGetCurrentRedirectCount() {
366375
returnredirectCount.incrementAndGet();
367376
}
368377

369-
publicvoidsetReaperFuture(FutureReaperreaperFuture) {
370-
cancelReaper();
371-
this.reaperFuture =reaperFuture;
378+
publicvoidsetTimeoutsHolder(TimeoutsHoldertimeoutsHolder) {
379+
this.timeoutsHolder =timeoutsHolder;
372380
}
373381

374382
publicbooleanisInAuth() {
@@ -411,6 +419,10 @@ public void setStreamWasAlreadyConsumed(boolean streamWasAlreadyConsumed) {
411419
publicvoidtouch() {
412420
touch.set(millisTime());
413421
}
422+
423+
publiclonggetLastTouch() {
424+
returntouch.get();
425+
}
414426

415427
@Override
416428
publicbooleangetAndSetWriteHeaders(booleanwriteHeaders) {
@@ -499,7 +511,7 @@ public String toString() {
499511
",\n\thttpHeaders=" +httpHeaders +//
500512
",\n\texEx=" +exEx +//
501513
",\n\tredirectCount=" +redirectCount +//
502-
",\n\treaperFuture=" +reaperFuture +//
514+
",\n\timeoutsHolder=" +timeoutsHolder +//
503515
",\n\tinAuth=" +inAuth +//
504516
",\n\tstatusReceived=" +statusReceived +//
505517
",\n\ttouch=" +touch +//

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp