36
36
import java .util .concurrent .Executor ;
37
37
import java .util .concurrent .Flow ;
38
38
import java .util .concurrent .TimeUnit ;
39
+
39
40
import org .springframework .http .HttpHeaders ;
40
41
import org .springframework .http .HttpMethod ;
41
42
import org .springframework .lang .Nullable ;
@@ -94,38 +95,25 @@ public URI getURI() {
94
95
@ Override
95
96
@ SuppressWarnings ("NullAway" )
96
97
protected ClientHttpResponse executeInternal (HttpHeaders headers , @ Nullable Body body ) throws IOException {
97
- HttpRequest request = buildRequest (headers , body );
98
- CompletableFuture <HttpResponse <InputStream >> responsefuture =
99
- this .httpClient .sendAsync (request , HttpResponse .BodyHandlers .ofInputStream ());
98
+ CompletableFuture <HttpResponse <InputStream >> responseFuture = null ;
100
99
try {
100
+ HttpRequest request = buildRequest (headers , body );
101
+ responseFuture = this .httpClient .sendAsync (request , HttpResponse .BodyHandlers .ofInputStream ());
102
+
101
103
if (this .timeout != null ) {
102
- CompletableFuture <Void > timeoutFuture = new CompletableFuture <Void >()
103
- .completeOnTimeout (null , this .timeout .toMillis (), TimeUnit .MILLISECONDS );
104
- timeoutFuture .thenRun (() -> {
105
- if (!responsefuture .cancel (true ) && !responsefuture .isCompletedExceptionally ()) {
106
- try {
107
- responsefuture .resultNow ().body ().close ();
108
- } catch (IOException ignored ) {}
109
- }
110
- });
111
- var response = responsefuture .get ();
112
- return new JdkClientHttpResponse (response .statusCode (), response .headers (), new FilterInputStream (response .body ()) {
113
-
114
- @ Override
115
- public void close () throws IOException {
116
- timeoutFuture .cancel (false );
117
- super .close ();
118
- }
119
- });
120
-
121
- } else {
122
- var response = responsefuture .get ();
123
- return new JdkClientHttpResponse (response .statusCode (), response .headers (), response .body ());
104
+ TimeoutHandler timeoutHandler = new TimeoutHandler (responseFuture , this .timeout );
105
+ HttpResponse <InputStream > response = responseFuture .get ();
106
+ InputStream inputStream = timeoutHandler .wrapInputStream (response );
107
+ return new JdkClientHttpResponse (response , inputStream );
108
+ }
109
+ else {
110
+ HttpResponse <InputStream > response = responseFuture .get ();
111
+ return new JdkClientHttpResponse (response , response .body ());
124
112
}
125
113
}
126
114
catch (InterruptedException ex ) {
127
115
Thread .currentThread ().interrupt ();
128
- responsefuture .cancel (true );
116
+ responseFuture .cancel (true );
129
117
throw new IOException ("Request was interrupted: " + ex .getMessage (), ex );
130
118
}
131
119
catch (ExecutionException ex ) {
@@ -149,7 +137,6 @@ else if (cause instanceof IOException ioEx) {
149
137
}
150
138
}
151
139
152
-
153
140
private HttpRequest buildRequest (HttpHeaders headers , @ Nullable Body body ) {
154
141
HttpRequest .Builder builder = HttpRequest .newBuilder ().uri (this .uri );
155
142
@@ -225,4 +212,52 @@ public ByteBuffer map(byte[] b, int off, int len) {
225
212
}
226
213
}
227
214
215
+
216
+ /**
217
+ * Temporary workaround to use instead of {@link HttpRequest.Builder#timeout(Duration)}
218
+ * until <a href="https://bugs.openjdk.org/browse/JDK-8258397">JDK-8258397</a>
219
+ * is fixed. Essentially, create a future wiht a timeout handler, and use it
220
+ * to close the response.
221
+ * @see <a href="https://mail.openjdk.org/pipermail/net-dev/2021-October/016672.html">OpenJDK discussion thread</a>
222
+ */
223
+ private static final class TimeoutHandler {
224
+
225
+ private final CompletableFuture <Void > timeoutFuture ;
226
+
227
+ private TimeoutHandler (CompletableFuture <HttpResponse <InputStream >> future , Duration timeout ) {
228
+
229
+ this .timeoutFuture = new CompletableFuture <Void >()
230
+ .completeOnTimeout (null , timeout .toMillis (), TimeUnit .MILLISECONDS );
231
+
232
+ this .timeoutFuture .thenRun (() -> {
233
+ if (future .cancel (true ) || future .isCompletedExceptionally () || !future .isDone ()) {
234
+ return ;
235
+ }
236
+ try {
237
+ future .get ().body ().close ();
238
+ }
239
+ catch (Exception ex ) {
240
+ // ignore
241
+ }
242
+ });
243
+
244
+ }
245
+
246
+ @ Nullable
247
+ public InputStream wrapInputStream (HttpResponse <InputStream > response ) {
248
+ InputStream body = response .body ();
249
+ if (body == null ) {
250
+ return body ;
251
+ }
252
+ return new FilterInputStream (body ) {
253
+
254
+ @ Override
255
+ public void close () throws IOException {
256
+ TimeoutHandler .this .timeoutFuture .cancel (false );
257
+ super .close ();
258
+ }
259
+ };
260
+ }
261
+ }
262
+
228
263
}
0 commit comments