Skip to content

Commit 9ba3d85

Browse files
zoewanggzoewangg
committed
Refactoring - moving the request exceution logic to a separate class and add tests
1 parent feebcd2 commit 9ba3d85

File tree

9 files changed

+564
-117
lines changed

9 files changed

+564
-117
lines changed

‎http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtAsyncHttpClient.java

Lines changed: 18 additions & 116 deletions
Original file line numberDiff line numberDiff line change
@@ -15,48 +15,37 @@
1515

1616
package software.amazon.awssdk.http.crt;
1717

18-
import static software.amazon.awssdk.utils.CollectionUtils.isNullOrEmpty;
19-
import static software.amazon.awssdk.utils.FunctionalUtils.invokeSafely;
2018
import static software.amazon.awssdk.utils.Validate.paramNotNull;
2119

22-
import java.io.IOException;
2320
import java.net.URI;
2421
import java.time.Duration;
25-
import java.util.ArrayList;
2622
import java.util.LinkedList;
27-
import java.util.List;
2823
import java.util.Map;
29-
import java.util.Optional;
3024
import java.util.concurrent.CompletableFuture;
3125
import java.util.concurrent.ConcurrentHashMap;
3226
import java.util.function.Consumer;
3327
import software.amazon.awssdk.annotations.SdkPublicApi;
3428
import software.amazon.awssdk.crt.CrtResource;
35-
import software.amazon.awssdk.crt.CrtRuntimeException;
3629
import software.amazon.awssdk.crt.http.HttpClientConnectionManager;
3730
import software.amazon.awssdk.crt.http.HttpClientConnectionManagerOptions;
38-
import software.amazon.awssdk.crt.http.HttpHeader;
3931
import software.amazon.awssdk.crt.http.HttpMonitoringOptions;
4032
import software.amazon.awssdk.crt.http.HttpProxyOptions;
41-
import software.amazon.awssdk.crt.http.HttpRequest;
4233
import software.amazon.awssdk.crt.io.ClientBootstrap;
4334
import software.amazon.awssdk.crt.io.EventLoopGroup;
4435
import software.amazon.awssdk.crt.io.HostResolver;
4536
import software.amazon.awssdk.crt.io.SocketOptions;
4637
import software.amazon.awssdk.crt.io.TlsCipherPreference;
4738
import software.amazon.awssdk.crt.io.TlsContext;
4839
import software.amazon.awssdk.crt.io.TlsContextOptions;
49-
import software.amazon.awssdk.http.Header;
5040
import software.amazon.awssdk.http.SdkHttpConfigurationOption;
51-
import software.amazon.awssdk.http.SdkHttpRequest;
5241
import software.amazon.awssdk.http.async.AsyncExecuteRequest;
5342
import software.amazon.awssdk.http.async.SdkAsyncHttpClient;
54-
import software.amazon.awssdk.http.crt.internal.AwsCrtAsyncHttpStreamAdapter;
43+
import software.amazon.awssdk.http.crt.internal.CrtRequestContext;
44+
import software.amazon.awssdk.http.crt.internal.CrtRequestExecutor;
5545
import software.amazon.awssdk.utils.AttributeMap;
5646
import software.amazon.awssdk.utils.IoUtils;
5747
import software.amazon.awssdk.utils.Logger;
5848
import software.amazon.awssdk.utils.Validate;
59-
import software.amazon.awssdk.utils.http.SdkHttpUtils;
6049

6150
/**
6251
* An implementation of {@link SdkAsyncHttpClient} that uses the AWS Common Runtime (CRT) Http Client to communicate with
@@ -79,7 +68,7 @@ public final class AwsCrtAsyncHttpClient implements SdkAsyncHttpClient {
7968
private final HttpProxyOptions proxyOptions;
8069
private final HttpMonitoringOptions monitoringOptions;
8170
private final long maxConnectionIdleInMilliseconds;
82-
private final int initialWindowSize;
71+
private final int readBufferSize;
8372
private final int maxConnectionsPerEndpoint;
8473
private boolean isClosed = false;
8574

@@ -88,7 +77,7 @@ private AwsCrtAsyncHttpClient(DefaultBuilder builder, AttributeMap config) {
8877

8978
Validate.isPositive(maxConns, "maxConns");
9079
Validate.notNull(builder.cipherPreference, "cipherPreference");
91-
Validate.isPositive(builder.initialWindowSize, "initialWindowSize");
80+
Validate.isPositive(builder.readBufferSize, "readBufferSize");
9281
Validate.notNull(builder.eventLoopGroup, "eventLoopGroup");
9382
Validate.notNull(builder.hostResolver, "hostResolver");
9483

@@ -102,12 +91,10 @@ private AwsCrtAsyncHttpClient(DefaultBuilder builder, AttributeMap config) {
10291
this.bootstrap = registerOwnedResource(clientBootstrap);
10392
this.socketOptions = registerOwnedResource(clientSocketOptions);
10493
this.tlsContext = registerOwnedResource(clientTlsContext);
105-
106-
this.initialWindowSize = builder.initialWindowSize;
94+
this.readBufferSize = builder.readBufferSize;
10795
this.maxConnectionsPerEndpoint = maxConns;
10896
this.monitoringOptions = revolveHttpMonitoringOptions(builder.connectionHealthChecksConfiguration);
10997
this.maxConnectionIdleInMilliseconds = config.get(SdkHttpConfigurationOption.CONNECTION_MAX_IDLE_TIMEOUT).toMillis();
110-
11198
this.proxyOptions = buildProxyOptions(builder.proxyConfiguration);
11299
}
113100
}
@@ -164,11 +151,6 @@ private <T extends CrtResource> T registerOwnedResource(T subresource) {
164151
return subresource;
165152
}
166153

167-
private static URI toUri(SdkHttpRequest sdkRequest) {
168-
return invokeSafely(() -> new URI(sdkRequest.protocol(), null, sdkRequest.host(), sdkRequest.port(),
169-
null, null, null));
170-
}
171-
172154
public static Builder builder() {
173155
return new DefaultBuilder();
174156
}
@@ -195,7 +177,7 @@ private HttpClientConnectionManager createConnectionPool(URI uri) {
195177
.withSocketOptions(socketOptions)
196178
.withTlsContext(tlsContext)
197179
.withUri(uri)
198-
.withWindowSize(initialWindowSize)
180+
.withWindowSize(readBufferSize)
199181
.withMaxConnections(maxConnectionsPerEndpoint)
200182
.withManualWindowManagement(true)
201183
.withProxyOptions(proxyOptions)
@@ -232,60 +214,6 @@ private HttpClientConnectionManager getOrCreateConnectionPool(URI uri) {
232214
}
233215
}
234216

235-
private List<HttpHeader> createHttpHeaderList(URI uri, AsyncExecuteRequest asyncRequest) {
236-
SdkHttpRequest sdkRequest = asyncRequest.request();
237-
// worst case we may add 3 more headers here
238-
List<HttpHeader> crtHeaderList = new ArrayList<>(sdkRequest.headers().size() + 3);
239-
240-
// Set Host Header if needed
241-
if (isNullOrEmpty(sdkRequest.headers().get(Header.HOST))) {
242-
crtHeaderList.add(new HttpHeader(Header.HOST, uri.getHost()));
243-
}
244-
245-
// Add Connection Keep Alive Header to reuse this Http Connection as long as possible
246-
if (isNullOrEmpty(sdkRequest.headers().get(Header.CONNECTION))) {
247-
crtHeaderList.add(new HttpHeader(Header.CONNECTION, Header.KEEP_ALIVE_VALUE));
248-
}
249-
250-
// Set Content-Length if needed
251-
Optional<Long> contentLength = asyncRequest.requestContentPublisher().contentLength();
252-
if (isNullOrEmpty(sdkRequest.headers().get(Header.CONTENT_LENGTH)) && contentLength.isPresent()) {
253-
crtHeaderList.add(new HttpHeader(Header.CONTENT_LENGTH, Long.toString(contentLength.get())));
254-
}
255-
256-
// Add the rest of the Headers
257-
for (Map.Entry<String, List<String>> headerList: sdkRequest.headers().entrySet()) {
258-
for (String val: headerList.getValue()) {
259-
HttpHeader h = new HttpHeader(headerList.getKey(), val);
260-
crtHeaderList.add(h);
261-
}
262-
}
263-
264-
return crtHeaderList;
265-
}
266-
267-
private HttpHeader[] asArray(List<HttpHeader> crtHeaderList) {
268-
return crtHeaderList.toArray(new HttpHeader[0]);
269-
}
270-
271-
private HttpRequest toCrtRequest(URI uri, AsyncExecuteRequest asyncRequest, AwsCrtAsyncHttpStreamAdapter crtToSdkAdapter) {
272-
SdkHttpRequest sdkRequest = asyncRequest.request();
273-
274-
String method = sdkRequest.method().name();
275-
String encodedPath = sdkRequest.encodedPath();
276-
if (encodedPath == null || encodedPath.length() == 0) {
277-
encodedPath = "/";
278-
}
279-
280-
String encodedQueryString = SdkHttpUtils.encodeAndFlattenQueryParameters(sdkRequest.rawQueryParameters())
281-
.map(value -> "?" + value)
282-
.orElse("");
283-
284-
HttpHeader[] crtHeaderArray = asArray(createHttpHeaderList(uri, asyncRequest));
285-
286-
return new HttpRequest(method, encodedPath + encodedQueryString, crtHeaderArray, crtToSdkAdapter);
287-
}
288-
289217
@Override
290218
public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
291219

@@ -294,8 +222,6 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
294222
paramNotNull(asyncRequest.requestContentPublisher(), "RequestContentPublisher");
295223
paramNotNull(asyncRequest.responseHandler(), "ResponseHandler");
296224

297-
URI uri = toUri(asyncRequest.request());
298-
299225
/*
300226
* See the note on getOrCreateConnectionPool()
301227
*
@@ -306,38 +232,14 @@ public CompletableFuture<Void> execute(AsyncExecuteRequest asyncRequest) {
306232
* we have a pool and no one can destroy it underneath us until we've finished submitting the
307233
* request)
308234
*/
309-
try (HttpClientConnectionManager crtConnPool = getOrCreateConnectionPool(uri)) {
310-
CompletableFuture<Void> requestFuture = new CompletableFuture<>();
311-
// When a Connection is ready from the Connection Pool, schedule the Request on the connection
312-
crtConnPool.acquireConnection()
313-
.whenComplete((crtConn, throwable) -> {
314-
// If we didn't get a connection for some reason, fail the request
315-
if (throwable != null) {
316-
try {
317-
asyncRequest.responseHandler().onError(throwable);
318-
} catch (Exception e) {
319-
log.warn(() -> "Exception while handling error", e);
320-
}
321-
requestFuture.completeExceptionally(new IOException(
322-
"Crt exception while acquiring connection", throwable));
323-
return;
324-
}
325-
AwsCrtAsyncHttpStreamAdapter crtToSdkAdapter =
326-
new AwsCrtAsyncHttpStreamAdapter(crtConn, requestFuture, asyncRequest, initialWindowSize);
327-
HttpRequest crtRequest = toCrtRequest(uri, asyncRequest, crtToSdkAdapter);
328-
// Submit the Request on this Connection
329-
invokeSafely(() -> {
330-
try {
331-
crtConn.makeRequest(crtRequest, crtToSdkAdapter).activate();
332-
} catch (IllegalStateException | CrtRuntimeException e) {
333-
log.error(() -> "Exception occurred when making the request", e);
334-
requestFuture.completeExceptionally(
335-
new IOException("Exception throw while submitting request to CRT http connection", e));
336-
}
337-
});
338-
});
339-
340-
return requestFuture;
235+
try (HttpClientConnectionManager crtConnPool = getOrCreateConnectionPool(asyncRequest.request().getUri())) {
236+
CrtRequestContext context = CrtRequestContext.builder()
237+
.crtConnPool(crtConnPool)
238+
.readBufferSize(readBufferSize)
239+
.request(asyncRequest)
240+
.build();
241+
242+
return new CrtRequestExecutor().execute(context);
341243
}
342244
}
343245

@@ -456,7 +358,7 @@ Builder connectionHealthChecksConfiguration(Consumer<ConnectionHealthChecksConfi
456358
private static final class DefaultBuilder implements Builder {
457359
private final AttributeMap.Builder standardOptions = AttributeMap.builder();
458360
private TlsCipherPreference cipherPreference = TlsCipherPreference.TLS_CIPHER_SYSTEM_DEFAULT;
459-
private int initialWindowSize = DEFAULT_STREAM_WINDOW_SIZE;
361+
private int readBufferSize = DEFAULT_STREAM_WINDOW_SIZE;
460362
private EventLoopGroup eventLoopGroup;
461363
private HostResolver hostResolver;
462364
private ProxyConfiguration proxyConfiguration;
@@ -495,9 +397,9 @@ public Builder tlsCipherPreference(TlsCipherPreference tlsCipherPreference) {
495397
}
496398

497399
@Override
498-
public Builder readBufferSize(int initialWindowSize) {
499-
Validate.isPositive(initialWindowSize, "initialWindowSize");
500-
this.initialWindowSize = initialWindowSize;
400+
public Builder readBufferSize(int readBufferSize) {
401+
Validate.isPositive(readBufferSize, "readBufferSize");
402+
this.readBufferSize = readBufferSize;
501403
return this;
502404
}
503405

‎http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/CrtRequestContext.java

Lines changed: 77 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,77 @@
1+
/*
2+
* Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License").
5+
* You may not use this file except in compliance with the License.
6+
* A copy of the License is located at
7+
*
8+
* http://aws.amazon.com/apache2.0
9+
*
10+
* or in the "license" file accompanying this file. This file is distributed
11+
* on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
12+
* express or implied. See the License for the specific language governing
13+
* permissions and limitations under the License.
14+
*/
15+
16+
package software.amazon.awssdk.http.crt.internal;
17+
18+
import software.amazon.awssdk.annotations.SdkInternalApi;
19+
import software.amazon.awssdk.crt.http.HttpClientConnectionManager;
20+
import software.amazon.awssdk.http.async.AsyncExecuteRequest;
21+
22+
@SdkInternalApi
23+
public final class CrtRequestContext {
24+
private final AsyncExecuteRequest request;
25+
private final int readBufferSize;
26+
private final HttpClientConnectionManager crtConnPool;
27+
28+
private CrtRequestContext(Builder builder) {
29+
this.request = builder.request;
30+
this.readBufferSize = builder.readBufferSize;
31+
this.crtConnPool = builder.crtConnPool;
32+
}
33+
34+
public static Builder builder() {
35+
return new Builder();
36+
}
37+
38+
public AsyncExecuteRequest sdkRequest() {
39+
return request;
40+
}
41+
42+
public int readBufferSize() {
43+
return readBufferSize;
44+
}
45+
46+
public HttpClientConnectionManager crtConnPool() {
47+
return crtConnPool;
48+
}
49+
50+
public static class Builder {
51+
private AsyncExecuteRequest request;
52+
private int readBufferSize;
53+
private HttpClientConnectionManager crtConnPool;
54+
55+
private Builder() {
56+
}
57+
58+
public Builder request(AsyncExecuteRequest request) {
59+
this.request = request;
60+
return this;
61+
}
62+
63+
public Builder readBufferSize(int readBufferSize) {
64+
this.readBufferSize = readBufferSize;
65+
return this;
66+
}
67+
68+
public Builder crtConnPool(HttpClientConnectionManager crtConnPool) {
69+
this.crtConnPool = crtConnPool;
70+
return this;
71+
}
72+
73+
public CrtRequestContext build() {
74+
return new CrtRequestContext(this);
75+
}
76+
}
77+
}

0 commit comments

Comments
 (0)