diff --git a/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkHttpClient.java b/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkHttpClient.java index 8f8a45ef498c..f2fcd2dad7d1 100644 --- a/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkHttpClient.java +++ b/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkHttpClient.java @@ -12,6 +12,7 @@ import com.azure.core.http.jdk.httpclient.implementation.InputStreamTimeoutResponseSubscriber; import com.azure.core.http.jdk.httpclient.implementation.JdkHttpResponseAsync; import com.azure.core.http.jdk.httpclient.implementation.JdkHttpResponseSync; +import com.azure.core.implementation.util.HttpUtils; import com.azure.core.util.Context; import com.azure.core.util.logging.ClientLogger; import reactor.core.publisher.Mono; @@ -31,8 +32,6 @@ */ class JdkHttpClient implements HttpClient { private static final ClientLogger LOGGER = new ClientLogger(JdkHttpClient.class); - private static final String AZURE_EAGERLY_READ_RESPONSE = "azure-eagerly-read-response"; - private static final String AZURE_IGNORE_RESPONSE_BODY = "azure-ignore-response-body"; private final java.net.http.HttpClient jdkHttpClient; @@ -79,8 +78,8 @@ public Mono send(HttpRequest request) { @Override public Mono send(HttpRequest request, Context context) { - boolean eagerlyReadResponse = (boolean) context.getData(AZURE_EAGERLY_READ_RESPONSE).orElse(false); - boolean ignoreResponseBody = (boolean) context.getData(AZURE_IGNORE_RESPONSE_BODY).orElse(false); + boolean eagerlyReadResponse = (boolean) context.getData(HttpUtils.AZURE_EAGERLY_READ_RESPONSE).orElse(false); + boolean ignoreResponseBody = (boolean) context.getData(HttpUtils.AZURE_IGNORE_RESPONSE_BODY).orElse(false); Mono jdkRequestMono = Mono.fromCallable(() -> toJdkHttpRequest(request, context)); @@ -111,8 +110,8 @@ public Mono send(HttpRequest request, Context context) { @Override public HttpResponse sendSync(HttpRequest request, Context context) { - boolean eagerlyReadResponse = (boolean) context.getData(AZURE_EAGERLY_READ_RESPONSE).orElse(false); - boolean ignoreResponseBody = (boolean) context.getData(AZURE_IGNORE_RESPONSE_BODY).orElse(false); + boolean eagerlyReadResponse = (boolean) context.getData(HttpUtils.AZURE_EAGERLY_READ_RESPONSE).orElse(false); + boolean ignoreResponseBody = (boolean) context.getData(HttpUtils.AZURE_IGNORE_RESPONSE_BODY).orElse(false); java.net.http.HttpRequest jdkRequest = toJdkHttpRequest(request, context); try { diff --git a/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkHttpClientBuilder.java b/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkHttpClientBuilder.java index 79e2da860819..c20b141ecb1f 100644 --- a/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkHttpClientBuilder.java +++ b/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkHttpClientBuilder.java @@ -25,11 +25,11 @@ import java.util.Set; import java.util.concurrent.Executor; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_READ_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT; -import static com.azure.core.util.CoreUtils.getDefaultTimeoutFromEnvironment; +import static com.azure.core.implementation.util.HttpUtils.getDefaultConnectTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultReadTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultResponseTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultWriteTimeout; +import static com.azure.core.implementation.util.HttpUtils.getTimeout; /** * Builder to configure and build an instance of the azure-core {@link HttpClient} type using the JDK HttpClient APIs, @@ -38,12 +38,6 @@ public class JdkHttpClientBuilder { private static final ClientLogger LOGGER = new ClientLogger(JdkHttpClientBuilder.class); - private static final Duration MINIMUM_TIMEOUT = Duration.ofMillis(1); - private static final Duration DEFAULT_CONNECTION_TIMEOUT; - private static final Duration DEFAULT_WRITE_TIMEOUT; - private static final Duration DEFAULT_RESPONSE_TIMEOUT; - private static final Duration DEFAULT_READ_TIMEOUT; - private static final String JAVA_HOME = System.getProperty("java.home"); private static final String JDK_HTTPCLIENT_ALLOW_RESTRICTED_HEADERS = "jdk.httpclient.allowRestrictedHeaders"; @@ -57,17 +51,6 @@ public class JdkHttpClientBuilder { static final Set DEFAULT_RESTRICTED_HEADERS; static { - Configuration configuration = Configuration.getGlobalConfiguration(); - - DEFAULT_CONNECTION_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, - PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT, Duration.ofSeconds(10), LOGGER); - DEFAULT_WRITE_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT, - Duration.ofSeconds(60), LOGGER); - DEFAULT_RESPONSE_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, - PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT, Duration.ofSeconds(60), LOGGER); - DEFAULT_READ_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, PROPERTY_AZURE_REQUEST_READ_TIMEOUT, - Duration.ofSeconds(60), LOGGER); - DEFAULT_RESTRICTED_HEADERS = Set.of("connection", "content-length", "expect", "host", "upgrade"); } @@ -248,11 +231,11 @@ public HttpClient build() { // Azure JDK http client supports HTTP 1.1 by default. httpClientBuilder.version(java.net.http.HttpClient.Version.HTTP_1_1); - httpClientBuilder = httpClientBuilder.connectTimeout(getTimeout(connectionTimeout, DEFAULT_CONNECTION_TIMEOUT)); + httpClientBuilder = httpClientBuilder.connectTimeout(getTimeout(connectionTimeout, getDefaultConnectTimeout())); - Duration writeTimeout = getTimeout(this.writeTimeout, DEFAULT_WRITE_TIMEOUT); - Duration responseTimeout = getTimeout(this.responseTimeout, DEFAULT_RESPONSE_TIMEOUT); - Duration readTimeout = getTimeout(this.readTimeout, DEFAULT_READ_TIMEOUT); + Duration writeTimeout = getTimeout(this.writeTimeout, getDefaultWriteTimeout()); + Duration responseTimeout = getTimeout(this.responseTimeout, getDefaultResponseTimeout()); + Duration readTimeout = getTimeout(this.readTimeout, getDefaultReadTimeout()); Configuration buildConfiguration = (configuration == null) ? Configuration.getGlobalConfiguration() : configuration; @@ -331,12 +314,4 @@ protected PasswordAuthentication getPasswordAuthentication() { return new PasswordAuthentication(this.userName, password.toCharArray()); } } - - private static Duration getTimeout(Duration configuredTimeout, Duration defaultTimeout) { - if (configuredTimeout == null) { - return defaultTimeout; - } - - return configuredTimeout.compareTo(MINIMUM_TIMEOUT) < 0 ? MINIMUM_TIMEOUT : configuredTimeout; - } } diff --git a/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/implementation/AzureJdkHttpRequest.java b/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/implementation/AzureJdkHttpRequest.java index 6cf58d3b4a36..4e833c2eec52 100644 --- a/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/implementation/AzureJdkHttpRequest.java +++ b/sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/implementation/AzureJdkHttpRequest.java @@ -4,6 +4,7 @@ import com.azure.core.http.HttpMethod; import com.azure.core.implementation.util.HttpHeadersAccessHelper; +import com.azure.core.implementation.util.HttpUtils; import com.azure.core.util.Context; import com.azure.core.util.Contexts; import com.azure.core.util.ProgressReporter; @@ -51,6 +52,9 @@ public AzureJdkHttpRequest(com.azure.core.http.HttpRequest azureCoreRequest, Con Set restrictedHeaders, ClientLogger logger, Duration writeTimeout, Duration responseTimeout) { HttpMethod method = azureCoreRequest.getHttpMethod(); ProgressReporter progressReporter = Contexts.with(context).getHttpRequestProgressReporter(); + responseTimeout = (Duration) context.getData(HttpUtils.AZURE_RESPONSE_TIMEOUT) + .filter(timeoutDuration -> timeoutDuration instanceof Duration) + .orElse(responseTimeout); this.method = method.toString(); this.bodyPublisher = (method == HttpMethod.GET || method == HttpMethod.HEAD) diff --git a/sdk/core/azure-core-http-jdk-httpclient/src/test/java/com/azure/core/http/jdk/httpclient/JdkHttpClientLocalTestServer.java b/sdk/core/azure-core-http-jdk-httpclient/src/test/java/com/azure/core/http/jdk/httpclient/JdkHttpClientLocalTestServer.java index 48ce978de02a..4cac959b2dcf 100644 --- a/sdk/core/azure-core-http-jdk-httpclient/src/test/java/com/azure/core/http/jdk/httpclient/JdkHttpClientLocalTestServer.java +++ b/sdk/core/azure-core-http-jdk-httpclient/src/test/java/com/azure/core/http/jdk/httpclient/JdkHttpClientLocalTestServer.java @@ -22,6 +22,8 @@ public final class JdkHttpClientLocalTestServer { private static volatile LocalTestServer proxyServer; private static final Semaphore PROXY_SERVER_SEMAPHORE = new Semaphore(1); + public static final String TIMEOUT = "/timeout"; + public static final byte[] SHORT_BODY = "hi there".getBytes(StandardCharsets.UTF_8); public static final byte[] LONG_BODY = createLongBody(); @@ -104,6 +106,16 @@ private static LocalTestServer initializeServer() { resp.getHttpOutput().write(SHORT_BODY, 5, 3); resp.getHttpOutput().flush(); resp.getHttpOutput().complete(Callback.NOOP); + } else if (get && TIMEOUT.equals(path)) { + try { + Thread.sleep(5000); + resp.setStatus(200); + resp.getHttpOutput().write(SHORT_BODY); + resp.getHttpOutput().flush(); + resp.getHttpOutput().complete(Callback.NOOP); + } catch (InterruptedException e) { + throw new ServletException(e); + } } else { throw new ServletException("Unexpected request: " + req.getMethod() + " " + path); } diff --git a/sdk/core/azure-core-http-jdk-httpclient/src/test/java/com/azure/core/http/jdk/httpclient/JdkHttpClientTests.java b/sdk/core/azure-core-http-jdk-httpclient/src/test/java/com/azure/core/http/jdk/httpclient/JdkHttpClientTests.java index 35503e58c998..88b0ef8dca21 100644 --- a/sdk/core/azure-core-http-jdk-httpclient/src/test/java/com/azure/core/http/jdk/httpclient/JdkHttpClientTests.java +++ b/sdk/core/azure-core-http-jdk-httpclient/src/test/java/com/azure/core/http/jdk/httpclient/JdkHttpClientTests.java @@ -8,6 +8,7 @@ import com.azure.core.http.HttpMethod; import com.azure.core.http.HttpRequest; import com.azure.core.http.HttpResponse; +import com.azure.core.implementation.util.HttpUtils; import com.azure.core.util.BinaryData; import com.azure.core.util.Context; import com.azure.core.util.Contexts; @@ -55,6 +56,7 @@ import static com.azure.core.http.jdk.httpclient.JdkHttpClientLocalTestServer.LONG_BODY; import static com.azure.core.http.jdk.httpclient.JdkHttpClientLocalTestServer.SHORT_BODY; +import static com.azure.core.http.jdk.httpclient.JdkHttpClientLocalTestServer.TIMEOUT; import static com.azure.core.test.utils.TestUtils.assertArraysEqual; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; @@ -469,6 +471,46 @@ public void slowEagerReadingTimesOutAsync() { .verify(Duration.ofSeconds(5)); } + @Test + public void perCallTimeout() { + HttpClient client = new JdkHttpClientBuilder().responseTimeout(Duration.ofSeconds(10)).build(); + + HttpRequest request = new HttpRequest(HttpMethod.GET, url(TIMEOUT)); + + // Verify a smaller timeout sent through Context times out the request. + StepVerifier.create(client.send(request, new Context(HttpUtils.AZURE_RESPONSE_TIMEOUT, Duration.ofSeconds(1)))) + .expectErrorMatches(e -> e instanceof HttpTimeoutException) + .verify(); + + // Then verify not setting a timeout through Context does not time out the request. + StepVerifier.create(client.send(request) + .flatMap(response -> Mono.zip(FluxUtil.collectBytesInByteBufferStream(response.getBody()), + Mono.just(response.getStatusCode())))) + .assertNext(tuple -> { + assertArraysEqual(SHORT_BODY, tuple.getT1()); + assertEquals(200, tuple.getT2()); + }) + .verifyComplete(); + } + + @Test + public void perCallTimeoutSync() { + HttpClient client = new JdkHttpClientBuilder().responseTimeout(Duration.ofSeconds(10)).build(); + + HttpRequest request = new HttpRequest(HttpMethod.GET, url(TIMEOUT)); + + // Verify a smaller timeout sent through Context times out the request. + RuntimeException ex = assertThrows(RuntimeException.class, + () -> client.sendSync(request, new Context(HttpUtils.AZURE_RESPONSE_TIMEOUT, Duration.ofSeconds(1)))); + assertInstanceOf(HttpTimeoutException.class, ex.getCause()); + + // Then verify not setting a timeout through Context does not time out the request. + try (HttpResponse response = client.sendSync(request, Context.NONE)) { + assertEquals(200, response.getStatusCode()); + assertArraysEqual(SHORT_BODY, response.getBodyAsBinaryData().toBytes()); + } + } + private static Mono getResponse(String path) { HttpClient client = new JdkHttpClientBuilder().build(); return doRequest(client, path); diff --git a/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClient.java b/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClient.java index 944b72081b40..e64b7b69fa1b 100644 --- a/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClient.java +++ b/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClient.java @@ -15,6 +15,7 @@ import com.azure.core.implementation.util.BinaryDataHelper; import com.azure.core.implementation.util.ByteArrayContent; import com.azure.core.implementation.util.FileContent; +import com.azure.core.implementation.util.HttpUtils; import com.azure.core.implementation.util.InputStreamContent; import com.azure.core.implementation.util.SerializableContent; import com.azure.core.implementation.util.StringContent; @@ -94,11 +95,6 @@ class NettyAsyncHttpClient implements HttpClient { private static final ClientLogger LOGGER = new ClientLogger(NettyAsyncHttpClient.class); private static final byte[] EMPTY_BYTES = new byte[0]; - private static final String AZURE_EAGERLY_READ_RESPONSE = "azure-eagerly-read-response"; - private static final String AZURE_IGNORE_RESPONSE_BODY = "azure-ignore-response-body"; - private static final String AZURE_RESPONSE_TIMEOUT = "azure-response-timeout"; - private static final String AZURE_EAGERLY_CONVERT_HEADERS = "azure-eagerly-convert-headers"; - final boolean disableBufferCopy; final boolean addProxyHandler; @@ -132,10 +128,11 @@ public Mono send(HttpRequest request, Context context) { Objects.requireNonNull(request.getUrl(), "'request.getUrl()' cannot be null."); Objects.requireNonNull(request.getUrl().getProtocol(), "'request.getUrl().getProtocol()' cannot be null."); - boolean eagerlyReadResponse = (boolean) context.getData(AZURE_EAGERLY_READ_RESPONSE).orElse(false); - boolean ignoreResponseBody = (boolean) context.getData(AZURE_IGNORE_RESPONSE_BODY).orElse(false); - boolean headersEagerlyConverted = (boolean) context.getData(AZURE_EAGERLY_CONVERT_HEADERS).orElse(false); - Long responseTimeout = context.getData(AZURE_RESPONSE_TIMEOUT) + boolean eagerlyReadResponse = (boolean) context.getData(HttpUtils.AZURE_EAGERLY_READ_RESPONSE).orElse(false); + boolean ignoreResponseBody = (boolean) context.getData(HttpUtils.AZURE_IGNORE_RESPONSE_BODY).orElse(false); + boolean headersEagerlyConverted + = (boolean) context.getData(HttpUtils.AZURE_EAGERLY_CONVERT_HEADERS).orElse(false); + Long responseTimeout = context.getData(HttpUtils.AZURE_RESPONSE_TIMEOUT) .filter(timeoutDuration -> timeoutDuration instanceof Duration) .map(timeoutDuration -> ((Duration) timeoutDuration).toMillis()) .orElse(null); diff --git a/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClientBuilder.java b/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClientBuilder.java index 75083d268df5..ccdadbe4e2c8 100644 --- a/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClientBuilder.java +++ b/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClientBuilder.java @@ -36,15 +36,14 @@ import java.nio.ByteBuffer; import java.time.Duration; import java.util.Objects; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; import java.util.regex.Pattern; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_READ_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT; -import static com.azure.core.util.CoreUtils.getDefaultTimeoutFromEnvironment; +import static com.azure.core.implementation.util.HttpUtils.getDefaultConnectTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultReadTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultResponseTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultWriteTimeout; +import static com.azure.core.implementation.util.HttpUtils.getTimeout; /** *

@@ -112,27 +111,10 @@ * @see NettyAsyncHttpClient */ public class NettyAsyncHttpClientBuilder { - private static final long MINIMUM_TIMEOUT = TimeUnit.MILLISECONDS.toMillis(1); - private static final long DEFAULT_CONNECT_TIMEOUT; - private static final long DEFAULT_WRITE_TIMEOUT; - private static final long DEFAULT_RESPONSE_TIMEOUT; - private static final long DEFAULT_READ_TIMEOUT; - // NettyAsyncHttpClientBuilder may be instantiated many times, use a static logger. private static final ClientLogger LOGGER = new ClientLogger(NettyAsyncHttpClientBuilder.class); static { - Configuration configuration = Configuration.getGlobalConfiguration(); - - DEFAULT_CONNECT_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, - PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT, Duration.ofSeconds(10), LOGGER).toMillis(); - DEFAULT_WRITE_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT, - Duration.ofSeconds(60), LOGGER).toMillis(); - DEFAULT_RESPONSE_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, - PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT, Duration.ofSeconds(60), LOGGER).toMillis(); - DEFAULT_READ_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, PROPERTY_AZURE_REQUEST_READ_TIMEOUT, - Duration.ofSeconds(60), LOGGER).toMillis(); - Utility.validateNettyVersions(); } @@ -205,9 +187,9 @@ public com.azure.core.http.HttpClient build() { addressResolverWasSetByBuilder = true; } - long writeTimeout = getTimeoutMillis(this.writeTimeout, DEFAULT_WRITE_TIMEOUT); - long responseTimeout = getTimeoutMillis(this.responseTimeout, DEFAULT_RESPONSE_TIMEOUT); - long readTimeout = getTimeoutMillis(this.readTimeout, DEFAULT_READ_TIMEOUT); + long writeTimeout = getTimeout(this.writeTimeout, getDefaultWriteTimeout()).toMillis(); + long responseTimeout = getTimeout(this.responseTimeout, getDefaultResponseTimeout()).toMillis(); + long readTimeout = getTimeout(this.readTimeout, getDefaultReadTimeout()).toMillis(); // Get the initial HttpResponseDecoderSpec and update it. // .httpResponseDecoder passes a new HttpResponseDecoderSpec and any existing configuration should be updated @@ -215,7 +197,7 @@ public com.azure.core.http.HttpClient build() { HttpResponseDecoderSpec initialSpec = nettyHttpClient.configuration().decoder(); nettyHttpClient = nettyHttpClient.port(port) .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, - (int) getTimeoutMillis(connectTimeout, DEFAULT_CONNECT_TIMEOUT)) + (int) getTimeout(connectTimeout, getDefaultConnectTimeout()).toMillis()) // TODO (alzimmer): What does validating HTTP response headers get us? .httpResponseDecoder(httpResponseDecoderSpec -> initialSpec.validateHeaders(false)) .doOnRequest( @@ -562,27 +544,6 @@ private static ProxyProvider.Proxy toReactorNettyProxyType(ProxyOptions.Type azu } } - /* - * Returns the timeout in milliseconds to use based on the passed Duration and default timeout. - * - * If the timeout is {@code null} the default timeout will be used. If the timeout is less than or equal to zero - * no timeout will be used. If the timeout is less than one millisecond a timeout of one millisecond will be used. - */ - static long getTimeoutMillis(Duration configuredTimeout, long defaultTimeout) { - // Timeout is null, use the default timeout. - if (configuredTimeout == null) { - return defaultTimeout; - } - - // Timeout is less than or equal to zero, return no timeout. - if (configuredTimeout.isZero() || configuredTimeout.isNegative()) { - return 0; - } - - // Return the maximum of the timeout period and the minimum allowed timeout period. - return Math.max(configuredTimeout.toMillis(), MINIMUM_TIMEOUT); - } - private static boolean shouldApplyProxy(SocketAddress socketAddress, Pattern nonProxyHostsPattern) { if (nonProxyHostsPattern == null) { return true; diff --git a/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyAsyncHttpClientBuilderTests.java b/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyAsyncHttpClientBuilderTests.java index 85abd25bf71e..44e3e9ed1cb2 100644 --- a/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyAsyncHttpClientBuilderTests.java +++ b/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyAsyncHttpClientBuilderTests.java @@ -46,6 +46,7 @@ import static com.azure.core.http.netty.implementation.NettyHttpClientLocalTestServer.DEFAULT_PATH; import static com.azure.core.http.netty.implementation.NettyHttpClientLocalTestServer.PREBUILT_CLIENT_PATH; +import static com.azure.core.implementation.util.HttpUtils.getTimeout; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertInstanceOf; @@ -544,7 +545,7 @@ public void buildEventLoopClient() { @ParameterizedTest @MethodSource("getTimeoutMillisSupplier") public void getTimeoutMillis(Duration timeout, long expected) { - assertEquals(expected, NettyAsyncHttpClientBuilder.getTimeoutMillis(timeout, 60000)); + assertEquals(expected, getTimeout(timeout, Duration.ofMillis(60000)).toMillis()); } private static Stream getTimeoutMillisSupplier() { diff --git a/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyAsyncHttpClientTests.java b/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyAsyncHttpClientTests.java index 59e5a5dcd781..173061698aae 100644 --- a/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyAsyncHttpClientTests.java +++ b/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyAsyncHttpClientTests.java @@ -18,7 +18,7 @@ import com.azure.core.http.netty.implementation.NettyHttpClientLocalTestServer; import com.azure.core.http.policy.FixedDelay; import com.azure.core.http.policy.RetryPolicy; -import com.azure.core.test.utils.TestUtils; +import com.azure.core.implementation.util.HttpUtils; import com.azure.core.util.BinaryData; import com.azure.core.util.Context; import com.azure.core.util.Contexts; @@ -70,6 +70,7 @@ import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -89,9 +90,12 @@ import static com.azure.core.http.netty.implementation.NettyHttpClientLocalTestServer.SHORT_POST_BODY_PATH; import static com.azure.core.http.netty.implementation.NettyHttpClientLocalTestServer.SHORT_POST_BODY_WITH_VALIDATION_PATH; import static com.azure.core.http.netty.implementation.NettyHttpClientLocalTestServer.TEST_HEADER; +import static com.azure.core.http.netty.implementation.NettyHttpClientLocalTestServer.TIMEOUT; +import static com.azure.core.test.utils.TestUtils.assertArraysEqual; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertLinesMatch; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; @@ -237,7 +241,7 @@ public void testConcurrentRequests() { .flatMap(response -> Mono.using(() -> response, HttpResponse::getBodyAsByteArray, HttpResponse::close)); StepVerifier.create(responses).thenConsumeWhile(response -> { - TestUtils.assertArraysEqual(LONG_BODY, response); + assertArraysEqual(LONG_BODY, response); return true; }).expectComplete().verify(Duration.ofSeconds(60)); } @@ -254,7 +258,7 @@ public void testConcurrentRequestsSync() throws InterruptedException { requests.add(() -> { try (HttpResponse response = doRequestSync(client, "/long")) { byte[] body = response.getBodyAsBinaryData().toBytes(); - TestUtils.assertArraysEqual(LONG_BODY, body); + assertArraysEqual(LONG_BODY, body); return null; } }); @@ -591,6 +595,46 @@ public void httpClientWithResolverUsesConfiguredResolverWithProxy() { } } + @Test + public void perCallTimeout() { + HttpClient client = new NettyAsyncHttpClientBuilder().responseTimeout(Duration.ofSeconds(10)).build(); + + HttpRequest request = new HttpRequest(HttpMethod.GET, url(TIMEOUT)); + + // Verify a smaller timeout sent through Context times out the request. + StepVerifier.create(client.send(request, new Context(HttpUtils.AZURE_RESPONSE_TIMEOUT, Duration.ofSeconds(1)))) + .expectErrorMatches(e -> e instanceof TimeoutException) + .verify(); + + // Then verify not setting a timeout through Context does not time out the request. + StepVerifier.create(client.send(request) + .flatMap(response -> Mono.zip(FluxUtil.collectBytesInByteBufferStream(response.getBody()), + Mono.just(response.getStatusCode())))) + .assertNext(tuple -> { + assertArraysEqual(SHORT_BODY, tuple.getT1()); + assertEquals(200, tuple.getT2()); + }) + .verifyComplete(); + } + + @Test + public void perCallTimeoutSync() { + HttpClient client = new NettyAsyncHttpClientBuilder().responseTimeout(Duration.ofSeconds(10)).build(); + + HttpRequest request = new HttpRequest(HttpMethod.GET, url(TIMEOUT)); + + // Verify a smaller timeout sent through Context times out the request. + RuntimeException ex = assertThrows(RuntimeException.class, + () -> client.sendSync(request, new Context(HttpUtils.AZURE_RESPONSE_TIMEOUT, Duration.ofSeconds(1)))); + assertInstanceOf(TimeoutException.class, ex.getCause()); + + // Then verify not setting a timeout through Context does not time out the request. + try (HttpResponse response = client.sendSync(request, Context.NONE)) { + assertArraysEqual(SHORT_BODY, response.getBodyAsBinaryData().toBytes()); + assertEquals(200, response.getStatusCode()); + } + } + private static Stream requestHeaderSupplier() { return Stream.of(Arguments.of(null, NULL_REPLACEMENT), Arguments.of("", ""), Arguments.of("aValue", "aValue")); } diff --git a/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/implementation/NettyHttpClientLocalTestServer.java b/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/implementation/NettyHttpClientLocalTestServer.java index 0f8c8c52ba0b..35df2c17338b 100644 --- a/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/implementation/NettyHttpClientLocalTestServer.java +++ b/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/implementation/NettyHttpClientLocalTestServer.java @@ -36,6 +36,7 @@ public final class NettyHttpClientLocalTestServer { public static final String EXPECTED_HEADER = "userAgent"; public static final String RETURN_HEADERS_AS_IS_PATH = "/returnHeadersAsIs"; public static final String PROXY_TO_ADDRESS = "/proxyToAddress"; + public static final String TIMEOUT = "/timeout"; public static final byte[] SHORT_BODY = "hi there".getBytes(StandardCharsets.UTF_8); public static final byte[] LONG_BODY = createLongBody(); @@ -134,6 +135,16 @@ private static LocalTestServer initializeServer() { resp.getHttpOutput().write("I'm a teapot".getBytes(StandardCharsets.UTF_8)); resp.getHttpOutput().flush(); resp.getHttpOutput().complete(Callback.NOOP); + } else if (get && TIMEOUT.equals(path)) { + try { + Thread.sleep(5000); + resp.setStatus(200); + resp.getHttpOutput().write(SHORT_BODY); + resp.getHttpOutput().flush(); + resp.getHttpOutput().complete(Callback.NOOP); + } catch (InterruptedException e) { + throw new ServletException(e); + } } else { throw new ServletException("Unexpected request: " + req.getMethod() + " " + path); } diff --git a/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClient.java b/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClient.java index 0474465ff1ed..ed142df50ca4 100644 --- a/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClient.java +++ b/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClient.java @@ -10,20 +10,25 @@ import com.azure.core.http.HttpMethod; import com.azure.core.http.HttpRequest; import com.azure.core.http.HttpResponse; +import com.azure.core.http.ProxyOptions; import com.azure.core.http.okhttp.implementation.BinaryDataRequestBody; import com.azure.core.http.okhttp.implementation.OkHttpAsyncBufferedResponse; import com.azure.core.http.okhttp.implementation.OkHttpAsyncResponse; import com.azure.core.http.okhttp.implementation.OkHttpFluxRequestBody; import com.azure.core.http.okhttp.implementation.OkHttpProgressReportingRequestBody; +import com.azure.core.http.okhttp.implementation.PerCallTimeoutCall; +import com.azure.core.http.okhttp.implementation.ResponseTimeoutListenerFactory; import com.azure.core.implementation.util.BinaryDataContent; import com.azure.core.implementation.util.BinaryDataHelper; import com.azure.core.implementation.util.FluxByteBufferContent; +import com.azure.core.implementation.util.HttpUtils; import com.azure.core.util.BinaryData; import com.azure.core.util.Context; import com.azure.core.util.Contexts; import com.azure.core.util.ProgressReporter; import com.azure.core.util.logging.ClientLogger; import okhttp3.Call; +import okhttp3.EventListener; import okhttp3.MediaType; import okhttp3.OkHttpClient; import okhttp3.Request; @@ -34,7 +39,9 @@ import reactor.core.publisher.MonoSink; import java.io.IOException; +import java.io.InterruptedIOException; import java.io.UncheckedIOException; +import java.time.Duration; /** * This class provides a OkHttp-based implementation for the {@link HttpClient} interface. Creating an instance of this @@ -70,14 +77,15 @@ class OkHttpAsyncHttpClient implements HttpClient { private static final byte[] EMPTY_BODY = new byte[0]; private static final RequestBody EMPTY_REQUEST_BODY = RequestBody.create(EMPTY_BODY); - private static final String AZURE_EAGERLY_READ_RESPONSE = "azure-eagerly-read-response"; - private static final String AZURE_IGNORE_RESPONSE_BODY = "azure-ignore-response-body"; - private static final String AZURE_EAGERLY_CONVERT_HEADERS = "azure-eagerly-convert-headers"; - final OkHttpClient httpClient; - OkHttpAsyncHttpClient(OkHttpClient httpClient) { - this.httpClient = httpClient; + private final Duration responseTimeout; + + OkHttpAsyncHttpClient(OkHttpClient httpClient, Duration responseTimeout) { + EventListener.Factory factory = httpClient.eventListenerFactory(); + this.httpClient + = httpClient.newBuilder().eventListenerFactory(new ResponseTimeoutListenerFactory(factory)).build(); + this.responseTimeout = responseTimeout; } @Override @@ -87,9 +95,13 @@ public Mono send(HttpRequest request) { @Override public Mono send(HttpRequest request, Context context) { - boolean eagerlyReadResponse = (boolean) context.getData(AZURE_EAGERLY_READ_RESPONSE).orElse(false); - boolean ignoreResponseBody = (boolean) context.getData(AZURE_IGNORE_RESPONSE_BODY).orElse(false); - boolean eagerlyConvertHeaders = (boolean) context.getData(AZURE_EAGERLY_CONVERT_HEADERS).orElse(false); + boolean eagerlyReadResponse = (boolean) context.getData(HttpUtils.AZURE_EAGERLY_READ_RESPONSE).orElse(false); + boolean ignoreResponseBody = (boolean) context.getData(HttpUtils.AZURE_IGNORE_RESPONSE_BODY).orElse(false); + boolean eagerlyConvertHeaders + = (boolean) context.getData(HttpUtils.AZURE_EAGERLY_CONVERT_HEADERS).orElse(false); + Duration perCallTimeout = (Duration) context.getData(HttpUtils.AZURE_RESPONSE_TIMEOUT) + .filter(timeoutDuration -> timeoutDuration instanceof Duration) + .orElse(responseTimeout); ProgressReporter progressReporter = Contexts.with(context).getHttpRequestProgressReporter(); @@ -106,47 +118,84 @@ public Mono send(HttpRequest request, Context context) { // 3. If Flux asynchronous then subscribe does not block caller thread // but block on the thread backing flux. This ignore any subscribeOn applied to send(r) // - Mono.fromCallable(() -> toOkHttpRequest(request, progressReporter)).subscribe(okHttpRequest -> { - try { - Call call = httpClient.newCall(okHttpRequest); - call.enqueue(new OkHttpCallback(sink, request, eagerlyReadResponse, ignoreResponseBody, - eagerlyConvertHeaders)); - sink.onCancel(call::cancel); - } catch (Exception ex) { - sink.error(ex); - } - }, sink::error); + Mono.fromCallable(() -> toOkHttpRequest(request, progressReporter, perCallTimeout)) + .subscribe(okHttpRequest -> { + try { + Call call = httpClient.newCall(okHttpRequest); + call.enqueue(new OkHttpCallback(sink, request, eagerlyReadResponse, ignoreResponseBody, + eagerlyConvertHeaders)); + sink.onCancel(call::cancel); + } catch (Exception ex) { + sink.error(ex); + } + }, sink::error); })); } @Override public HttpResponse sendSync(HttpRequest request, Context context) { - boolean eagerlyReadResponse = (boolean) context.getData(AZURE_EAGERLY_READ_RESPONSE).orElse(false); - boolean ignoreResponseBody = (boolean) context.getData(AZURE_IGNORE_RESPONSE_BODY).orElse(false); - boolean eagerlyConvertHeaders = (boolean) context.getData(AZURE_EAGERLY_CONVERT_HEADERS).orElse(false); + boolean eagerlyReadResponse = (boolean) context.getData(HttpUtils.AZURE_EAGERLY_READ_RESPONSE).orElse(false); + boolean ignoreResponseBody = (boolean) context.getData(HttpUtils.AZURE_IGNORE_RESPONSE_BODY).orElse(false); + boolean eagerlyConvertHeaders + = (boolean) context.getData(HttpUtils.AZURE_EAGERLY_CONVERT_HEADERS).orElse(false); + Duration perCallTimeout = (Duration) context.getData(HttpUtils.AZURE_RESPONSE_TIMEOUT) + .filter(timeoutDuration -> timeoutDuration instanceof Duration) + .orElse(responseTimeout); ProgressReporter progressReporter = Contexts.with(context).getHttpRequestProgressReporter(); - Request okHttpRequest = toOkHttpRequest(request, progressReporter); + Request okHttpRequest = toOkHttpRequest(request, progressReporter, perCallTimeout); + Call call = null; try { - Response okHttpResponse = httpClient.newCall(okHttpRequest).execute(); + call = httpClient.newCall(okHttpRequest); + Response okHttpResponse = call.execute(); return toHttpResponse(request, okHttpResponse, eagerlyReadResponse, ignoreResponseBody, eagerlyConvertHeaders); } catch (IOException e) { - throw LOGGER.logExceptionAsError(new UncheckedIOException(e)); + throw LOGGER.logExceptionAsError(new UncheckedIOException(mapIOException(e, call))); } } + /** + * Current design for response timeout uses call cancellation which throws an IOException with message "canceled". + * This isn't what we want, we want an InterruptedIOException with message "timeout". Use information stored on the + * call to determine if the IOException should be mapped to an InterruptedIOException. + * + * @param e the IOException to map + * @param call the Call to associate with the new IOException + * @return the new IOException + */ + private static IOException mapIOException(IOException e, Call call) { + if (call == null) { + return e; + } + + PerCallTimeoutCall perCallTimeoutCall = call.request().tag(PerCallTimeoutCall.class); + if (perCallTimeoutCall != null && perCallTimeoutCall.isTimedOut()) { + InterruptedIOException i = new InterruptedIOException("timedout"); + i.addSuppressed(e); + return i; + } + + return e; + } + /** * Converts the given azure-core request to okhttp request. * * @param request the azure-core request * @param progressReporter the {@link ProgressReporter}. Can be null. + * @param perCallTimeout the per call timeout * @return the okhttp request */ - private okhttp3.Request toOkHttpRequest(HttpRequest request, ProgressReporter progressReporter) { + private okhttp3.Request toOkHttpRequest(HttpRequest request, ProgressReporter progressReporter, + Duration perCallTimeout) { Request.Builder requestBuilder = new Request.Builder().url(request.getUrl()); + if (perCallTimeout != null) { + requestBuilder.tag(PerCallTimeoutCall.class, new PerCallTimeoutCall(perCallTimeout.toMillis())); + } + if (request.getHeaders() != null) { for (HttpHeader hdr : request.getHeaders()) { // OkHttp allows for headers with multiple values, but it treats them as separate headers, @@ -263,9 +312,14 @@ public void onFailure(okhttp3.Call call, IOException e) { if (e.getSuppressed().length == 1) { // Propagate suppressed exception when there is one. // This happens when body emission fails in the middle. - sink.error(e.getSuppressed()[0]); + Throwable suppressed = e.getSuppressed()[0]; + if (suppressed instanceof IOException) { + sink.error(mapIOException((IOException) suppressed, call)); + } else { + sink.error(suppressed); + } } else { - sink.error(e); + sink.error(mapIOException(e, call)); } } diff --git a/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClientBuilder.java b/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClientBuilder.java index f34406773f5c..ada7f41fd95b 100644 --- a/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClientBuilder.java +++ b/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClientBuilder.java @@ -4,10 +4,12 @@ package com.azure.core.http.okhttp; import com.azure.core.http.HttpClient; +import com.azure.core.http.HttpRequest; import com.azure.core.http.ProxyOptions; import com.azure.core.http.okhttp.implementation.OkHttpProxySelector; import com.azure.core.http.okhttp.implementation.ProxyAuthenticator; import com.azure.core.util.Configuration; +import com.azure.core.util.Context; import com.azure.core.util.logging.ClientLogger; import okhttp3.ConnectionPool; import okhttp3.Dispatcher; @@ -19,16 +21,17 @@ import java.util.List; import java.util.Objects; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_READ_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT; -import static com.azure.core.util.CoreUtils.getDefaultTimeoutFromEnvironment; +import static com.azure.core.implementation.util.HttpUtils.getDefaultConnectTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultReadTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultResponseTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultWriteTimeout; +import static com.azure.core.implementation.util.HttpUtils.getTimeout; /** * Builder class responsible for creating instances of {@link com.azure.core.http.HttpClient} backed by OkHttp. - * The client built from this builder can support sending requests synchronously and asynchronously. - * Use {@link com.azure.core.http.HttpClient#sendSync(HttpRequest, Context)} to send the provided request - * synchronously with contextual information. + * The client built from this builder can support sending requests synchronously and asynchronously. Use + * {@link HttpClient#sendSync(HttpRequest, Context)} to send the provided request synchronously with contextual + * information. * *

* Building a new HttpClient instance @@ -114,24 +117,9 @@ public class OkHttpAsyncHttpClientBuilder { private final okhttp3.OkHttpClient okHttpClient; - private static final Duration MINIMUM_TIMEOUT = Duration.ofMillis(1); - private static final Duration DEFAULT_CONNECT_TIMEOUT; - private static final Duration DEFAULT_WRITE_TIMEOUT; - private static final Duration DEFAULT_READ_TIMEOUT; - - static { - ClientLogger logger = new ClientLogger(OkHttpAsyncHttpClientBuilder.class); - Configuration configuration = Configuration.getGlobalConfiguration(); - DEFAULT_CONNECT_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, - PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT, Duration.ofSeconds(10), logger); - DEFAULT_WRITE_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT, - Duration.ofSeconds(60), logger); - DEFAULT_READ_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, PROPERTY_AZURE_REQUEST_READ_TIMEOUT, - Duration.ofSeconds(60), logger); - } - private List networkInterceptors = new ArrayList<>(); private Duration readTimeout; + private Duration responseTimeout; private Duration writeTimeout; private Duration connectionTimeout; private Duration callTimeout; @@ -199,11 +187,31 @@ public OkHttpAsyncHttpClientBuilder networkInterceptors(List networ * @see OkHttpClient.Builder#readTimeout(Duration) */ public OkHttpAsyncHttpClientBuilder readTimeout(Duration readTimeout) { - // setReadTimeout can be null this.readTimeout = readTimeout; return this; } + /** + * Sets the response timeout duration used when waiting for a server to reply. + *

+ * The response timeout begins once the request write completes and finishes once the first response read is + * triggered when the server response is received. + *

+ * If {@code responseTimeout} is null either {@link Configuration#PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT} or a + * 60-second timeout will be used, if it is a {@link Duration} less than or equal to zero then no timeout will be + * applied to the response. When applying the timeout the greatest of one millisecond and the value of {@code + * responseTimeout} will be used. + *

+ * Given OkHttp doesn't have an equivalent timeout for just responses, this is handled manually. + * + * @param responseTimeout Response timeout duration. + * @return The updated OkHttpAsyncHttpClientBuilder object. + */ + public OkHttpAsyncHttpClientBuilder responseTimeout(Duration responseTimeout) { + this.responseTimeout = responseTimeout; + return this; + } + /** * Sets the writing timeout for a request to be sent. *

@@ -355,9 +363,9 @@ public HttpClient build() { } // Configure operation timeouts. - httpClientBuilder = httpClientBuilder.connectTimeout(getTimeout(connectionTimeout, DEFAULT_CONNECT_TIMEOUT)) - .writeTimeout(getTimeout(writeTimeout, DEFAULT_WRITE_TIMEOUT)) - .readTimeout(getTimeout(readTimeout, DEFAULT_READ_TIMEOUT)); + httpClientBuilder = httpClientBuilder.connectTimeout(getTimeout(connectionTimeout, getDefaultConnectTimeout())) + .writeTimeout(getTimeout(writeTimeout, getDefaultWriteTimeout())) + .readTimeout(getTimeout(readTimeout, getDefaultReadTimeout())); if (callTimeout != null) { // Call timeout is disabled by default. @@ -397,31 +405,7 @@ public HttpClient build() { // Set the followRedirects property. httpClientBuilder.followRedirects(this.followRedirects); - return new OkHttpAsyncHttpClient(httpClientBuilder.build()); - } - - /* - * Returns the timeout in milliseconds to use based on the passed Duration and default timeout. - * - * If the timeout is {@code null} the default timeout will be used. If the timeout is less than or equal to zero - * no timeout will be used. If the timeout is less than one millisecond a timeout of one millisecond will be used. - */ - static Duration getTimeout(Duration configuredTimeout, Duration defaultTimeout) { - // Timeout is null, use the default timeout. - if (configuredTimeout == null) { - return defaultTimeout; - } - - // Timeout is less than or equal to zero, return no timeout. - if (configuredTimeout.isZero() || configuredTimeout.isNegative()) { - return Duration.ZERO; - } - - // Return the maximum of the timeout period and the minimum allowed timeout period. - if (configuredTimeout.compareTo(MINIMUM_TIMEOUT) < 0) { - return MINIMUM_TIMEOUT; - } else { - return configuredTimeout; - } + return new OkHttpAsyncHttpClient(httpClientBuilder.build(), + getTimeout(responseTimeout, getDefaultResponseTimeout())); } } diff --git a/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/implementation/PerCallTimeoutCall.java b/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/implementation/PerCallTimeoutCall.java new file mode 100644 index 000000000000..c37173dc47cf --- /dev/null +++ b/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/implementation/PerCallTimeoutCall.java @@ -0,0 +1,92 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.core.http.okhttp.implementation; + +import okhttp3.Call; + +import java.util.Timer; +import java.util.TimerTask; +import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; + +/** + * Implementation of {@link Call} which applies a per-call response timeout to the call. + */ +public final class PerCallTimeoutCall { + // Singleton timer to schedule timeout tasks. + // TODO (alzimmer): Make sure one thread is sufficient for all timeout tasks. + private static final Timer TIMER = new Timer("azure-okhttp-response-timeout-tracker", true); + + private final long perCallTimeout; + + private volatile boolean timedOut; + + private static final AtomicReferenceFieldUpdater CURRENT_TIMEOUT_UPDATER + = AtomicReferenceFieldUpdater.newUpdater(PerCallTimeoutCall.class, TimerTask.class, "currentTimeout"); + private volatile TimerTask currentTimeout; + + /** + * Creates a new instance of PerCallTimeoutCall. + * + * @param perCallTimeout The per-call timeout to apply to the call. + */ + public PerCallTimeoutCall(long perCallTimeout) { + this.perCallTimeout = perCallTimeout; + } + + /** + * Starts the timer task to time out the call if it exceeds the per-call timeout. + *

+ * No timeout is applied if the per-call timeout is less than or equal to 0. + * + * @param call The call to apply the timeout to. + */ + public void beginPerCallTimeout(Call call) { + if (perCallTimeout > 0) { + TimerTask currentTimeout = new PerCallTimerTask(this, call); + + TIMER.schedule(currentTimeout, perCallTimeout); + TimerTask existing = CURRENT_TIMEOUT_UPDATER.getAndSet(this, currentTimeout); + if (existing != null) { + existing.cancel(); + } + } + } + + /** + * Cancels the per-call timeout task. + *

+ * Cancellations happen if the response returned before the timeout or if the call was cancelled externally. + */ + public void endPerCallTimeout() { + TimerTask currentTimeout = CURRENT_TIMEOUT_UPDATER.getAndSet(this, null); + if (currentTimeout != null) { + currentTimeout.cancel(); + } + } + + /** + * Returns whether the call timed out. + * + * @return Whether the call timed out. + */ + public boolean isTimedOut() { + return timedOut; + } + + private static final class PerCallTimerTask extends TimerTask { + private final PerCallTimeoutCall perCallTimeoutCall; + private final Call call; + + PerCallTimerTask(PerCallTimeoutCall perCallTimeoutCall, Call call) { + this.perCallTimeoutCall = perCallTimeoutCall; + this.call = call; + } + + @Override + public void run() { + // Set timeout first. + perCallTimeoutCall.timedOut = true; + call.cancel(); + } + } +} diff --git a/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/implementation/ResponseTimeoutListener.java b/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/implementation/ResponseTimeoutListener.java new file mode 100644 index 000000000000..fde446e419b7 --- /dev/null +++ b/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/implementation/ResponseTimeoutListener.java @@ -0,0 +1,220 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.core.http.okhttp.implementation; + +import okhttp3.Call; +import okhttp3.Connection; +import okhttp3.EventListener; +import okhttp3.Handshake; +import okhttp3.HttpUrl; +import okhttp3.Protocol; +import okhttp3.Request; +import okhttp3.Response; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Proxy; +import java.util.List; + +/** + * Implementation of {@link EventListener} that watches {@link #requestHeadersEnd(Call, Request)} and + * {@link #requestBodyEnd(Call, long)} to detect when a request has completed being sent and + * {@link #responseHeadersStart(Call)} and {@link #responseFailed(Call, IOException)} to detect when a response has + * started being received. This is used to determine a response timeout for the call. + *

+ * Both completing the request headers and the response body need to be watched as requests without a body will never + * trigger the request body events. + */ +public class ResponseTimeoutListener extends EventListener { + private final EventListener delegate; + + /** + * Creates a new instance of ResponseTimeoutListener. + * + * @param delegate The {@link EventListener} to delegate to. + */ + public ResponseTimeoutListener(EventListener delegate) { + this.delegate = delegate; + } + + @Override + public void requestHeadersEnd(Call call, Request request) { + PerCallTimeoutCall perCallTimeoutCall = request.tag(PerCallTimeoutCall.class); + if (perCallTimeoutCall != null) { + if (request.body() == null) { + // Start the per call timeout when the request headers have been sent if there isn't a body. + perCallTimeoutCall.beginPerCallTimeout(call); + } + } + + delegate.requestHeadersEnd(call, request); + } + + @Override + public void requestFailed(Call call, IOException ioe) { + PerCallTimeoutCall perCallTimeoutCall = call.request().tag(PerCallTimeoutCall.class); + if (perCallTimeoutCall != null) { + // End the per call timeout if the request fails. + perCallTimeoutCall.endPerCallTimeout(); + } + + delegate.requestFailed(call, ioe); + } + + @Override + public void requestBodyEnd(Call call, long byteCount) { + PerCallTimeoutCall perCallTimeoutCall = call.request().tag(PerCallTimeoutCall.class); + if (perCallTimeoutCall != null) { + // Start the per call timeout when the request body has been sent. + perCallTimeoutCall.beginPerCallTimeout(call); + } + delegate.requestBodyEnd(call, byteCount); + } + + @Override + public void responseHeadersStart(Call call) { + PerCallTimeoutCall perCallTimeoutCall = call.request().tag(PerCallTimeoutCall.class); + if (perCallTimeoutCall != null) { + // End the per call timeout when the response headers have started being received. + perCallTimeoutCall.endPerCallTimeout(); + } + delegate.responseHeadersStart(call); + } + + @Override + public void responseFailed(Call call, IOException ioe) { + PerCallTimeoutCall perCallTimeoutCall = call.request().tag(PerCallTimeoutCall.class); + if (perCallTimeoutCall != null) { + // End the per call timeout if the response fails. + perCallTimeoutCall.endPerCallTimeout(); + } + delegate.responseFailed(call, ioe); + } + + @Override + public void canceled(Call call) { + PerCallTimeoutCall perCallTimeoutCall = call.request().tag(PerCallTimeoutCall.class); + if (perCallTimeoutCall != null) { + // End the per call timeout if the call is cancelled. + perCallTimeoutCall.endPerCallTimeout(); + } + delegate.canceled(call); + } + + @Override + public void cacheConditionalHit(Call call, Response cachedResponse) { + delegate.cacheConditionalHit(call, cachedResponse); + } + + @Override + public void cacheHit(Call call, Response response) { + delegate.cacheHit(call, response); + } + + @Override + public void cacheMiss(Call call) { + delegate.cacheMiss(call); + } + + @Override + public void callEnd(Call call) { + delegate.callEnd(call); + } + + @Override + public void callFailed(Call call, IOException ioe) { + delegate.callFailed(call, ioe); + } + + @Override + public void callStart(Call call) { + delegate.callStart(call); + } + + @Override + public void connectEnd(Call call, InetSocketAddress inetSocketAddress, Proxy proxy, Protocol protocol) { + delegate.connectEnd(call, inetSocketAddress, proxy, protocol); + } + + @Override + public void connectFailed(Call call, InetSocketAddress inetSocketAddress, Proxy proxy, Protocol protocol, + IOException ioe) { + delegate.connectFailed(call, inetSocketAddress, proxy, protocol, ioe); + } + + @Override + public void connectStart(Call call, InetSocketAddress inetSocketAddress, Proxy proxy) { + delegate.connectStart(call, inetSocketAddress, proxy); + } + + @Override + public void connectionAcquired(Call call, Connection connection) { + delegate.connectionAcquired(call, connection); + } + + @Override + public void connectionReleased(Call call, Connection connection) { + delegate.connectionReleased(call, connection); + } + + @Override + public void dnsEnd(Call call, String domainName, List inetAddressList) { + delegate.dnsEnd(call, domainName, inetAddressList); + } + + @Override + public void dnsStart(Call call, String domainName) { + delegate.dnsStart(call, domainName); + } + + @Override + public void proxySelectEnd(Call call, HttpUrl url, List proxies) { + delegate.proxySelectEnd(call, url, proxies); + } + + @Override + public void proxySelectStart(Call call, HttpUrl url) { + delegate.proxySelectStart(call, url); + } + + @Override + public void requestHeadersStart(Call call) { + delegate.requestHeadersStart(call); + } + + @Override + public void requestBodyStart(Call call) { + delegate.requestBodyStart(call); + } + + @Override + public void responseBodyEnd(Call call, long byteCount) { + delegate.responseBodyEnd(call, byteCount); + } + + @Override + public void responseBodyStart(Call call) { + delegate.responseBodyStart(call); + } + + @Override + public void responseHeadersEnd(Call call, Response response) { + delegate.responseHeadersEnd(call, response); + } + + @Override + public void satisfactionFailure(Call call, Response response) { + delegate.satisfactionFailure(call, response); + } + + @Override + public void secureConnectEnd(Call call, Handshake handshake) { + delegate.secureConnectEnd(call, handshake); + } + + @Override + public void secureConnectStart(Call call) { + delegate.secureConnectStart(call); + } +} diff --git a/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/implementation/ResponseTimeoutListenerFactory.java b/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/implementation/ResponseTimeoutListenerFactory.java new file mode 100644 index 000000000000..c6b57b19933e --- /dev/null +++ b/sdk/core/azure-core-http-okhttp/src/main/java/com/azure/core/http/okhttp/implementation/ResponseTimeoutListenerFactory.java @@ -0,0 +1,26 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.core.http.okhttp.implementation; + +import okhttp3.Call; +import okhttp3.EventListener; + +/** + * Implementation of {@link EventListener.Factory} that creates {@link ResponseTimeoutListener} instances. + */ +public final class ResponseTimeoutListenerFactory implements EventListener.Factory { + private final EventListener.Factory delegate; + + /** + * Creates a new instance of ResponseTimeoutListenerFactory. + * @param delegate The delegate factory to create the listener. + */ + public ResponseTimeoutListenerFactory(EventListener.Factory delegate) { + this.delegate = delegate; + } + + @Override + public EventListener create(Call call) { + return new ResponseTimeoutListener(delegate.create(call)); + } +} diff --git a/sdk/core/azure-core-http-okhttp/src/test/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClientTests.java b/sdk/core/azure-core-http-okhttp/src/test/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClientTests.java index 1c0783c436d6..3ff9611c8a28 100644 --- a/sdk/core/azure-core-http-okhttp/src/test/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClientTests.java +++ b/sdk/core/azure-core-http-okhttp/src/test/java/com/azure/core/http/okhttp/OkHttpAsyncHttpClientTests.java @@ -10,7 +10,9 @@ import com.azure.core.http.HttpMethod; import com.azure.core.http.HttpRequest; import com.azure.core.http.HttpResponse; +import com.azure.core.implementation.util.HttpUtils; import com.azure.core.util.Context; +import com.azure.core.util.FluxUtil; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.parallel.Execution; @@ -23,6 +25,8 @@ import reactor.test.StepVerifierOptions; import java.io.IOException; +import java.io.InterruptedIOException; +import java.io.UncheckedIOException; import java.net.MalformedURLException; import java.net.URI; import java.net.URISyntaxException; @@ -40,10 +44,14 @@ import static com.azure.core.http.okhttp.OkHttpClientLocalTestServer.LONG_BODY; import static com.azure.core.http.okhttp.OkHttpClientLocalTestServer.RETURN_HEADERS_AS_IS_PATH; import static com.azure.core.http.okhttp.OkHttpClientLocalTestServer.SHORT_BODY; +import static com.azure.core.http.okhttp.OkHttpClientLocalTestServer.TIMEOUT; import static com.azure.core.http.okhttp.TestUtils.createQuietDispatcher; +import static com.azure.core.test.utils.TestUtils.assertArraysEqual; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertLinesMatch; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @Execution(ExecutionMode.SAME_THREAD) @@ -224,6 +232,46 @@ public void validateHeadersReturnAsIs() { .verify(Duration.ofSeconds(10)); } + @Test + public void perCallTimeout() { + HttpClient client = new OkHttpAsyncHttpClientBuilder().responseTimeout(Duration.ofSeconds(20)).build(); + + HttpRequest request = new HttpRequest(HttpMethod.GET, url(TIMEOUT)); + + // Verify a smaller timeout sent through Context times out the request. + StepVerifier.create(client.send(request, new Context(HttpUtils.AZURE_RESPONSE_TIMEOUT, Duration.ofSeconds(1)))) + .expectErrorMatches(e -> e instanceof InterruptedIOException) + .verify(); + + // Then verify not setting a timeout through Context does not time out the request. + StepVerifier.create(client.send(request) + .flatMap(response -> Mono.zip(FluxUtil.collectBytesInByteBufferStream(response.getBody()), + Mono.just(response.getStatusCode())))) + .assertNext(tuple -> { + assertArraysEqual(SHORT_BODY, tuple.getT1()); + assertEquals(200, tuple.getT2()); + }) + .verifyComplete(); + } + + @Test + public void perCallTimeoutSync() { + HttpClient client = new OkHttpAsyncHttpClientBuilder().responseTimeout(Duration.ofSeconds(20)).build(); + + HttpRequest request = new HttpRequest(HttpMethod.GET, url(TIMEOUT)); + + // Verify a smaller timeout sent through Context times out the request. + UncheckedIOException ex = assertThrows(UncheckedIOException.class, + () -> client.sendSync(request, new Context(HttpUtils.AZURE_RESPONSE_TIMEOUT, Duration.ofSeconds(1)))); + assertInstanceOf(InterruptedIOException.class, ex.getCause()); + + // Then verify not setting a timeout through Context does not time out the request. + try (HttpResponse response = client.sendSync(request, Context.NONE)) { + assertEquals(200, response.getStatusCode()); + assertArraysEqual(SHORT_BODY, response.getBodyAsBinaryData().toBytes()); + } + } + private static Mono getResponse(String path) { HttpClient client = new OkHttpAsyncHttpClientBuilder().build(); return getResponse(client, path); diff --git a/sdk/core/azure-core-http-okhttp/src/test/java/com/azure/core/http/okhttp/OkHttpClientLocalTestServer.java b/sdk/core/azure-core-http-okhttp/src/test/java/com/azure/core/http/okhttp/OkHttpClientLocalTestServer.java index ad5a00e7ab52..cd1ec54f932a 100644 --- a/sdk/core/azure-core-http-okhttp/src/test/java/com/azure/core/http/okhttp/OkHttpClientLocalTestServer.java +++ b/sdk/core/azure-core-http-okhttp/src/test/java/com/azure/core/http/okhttp/OkHttpClientLocalTestServer.java @@ -28,6 +28,7 @@ public final class OkHttpClientLocalTestServer { public static final String DISPATCHER_PATH = "/dispatcher"; public static final String REDIRECT_PATH = "/redirect"; public static final String LOCATION_PATH = "/location"; + public static final String TIMEOUT = "/timeout"; public static final byte[] SHORT_BODY = "hi there".getBytes(StandardCharsets.UTF_8); public static final byte[] LONG_BODY = createLongBody(); @@ -119,6 +120,16 @@ private static LocalTestServer initializeServer() { }); } else if (get && "/connectionClose".equals(path)) { resp.getHttpChannel().getConnection().close(); + } else if (get && TIMEOUT.equals(path)) { + try { + Thread.sleep(5000); + resp.setStatus(200); + resp.getHttpOutput().write(SHORT_BODY); + resp.getHttpOutput().flush(); + resp.getHttpOutput().complete(Callback.NOOP); + } catch (InterruptedException e) { + throw new ServletException(e); + } } else { throw new ServletException("Unexpected request: " + req.getMethod() + " " + path); } diff --git a/sdk/core/azure-core-http-vertx/pom.xml b/sdk/core/azure-core-http-vertx/pom.xml index b077001bce30..7e2283c0db87 100644 --- a/sdk/core/azure-core-http-vertx/pom.xml +++ b/sdk/core/azure-core-http-vertx/pom.xml @@ -96,12 +96,6 @@ 4.5.7 - - io.vertx - vertx-reactive-streams - 4.5.7 - - com.azure diff --git a/sdk/core/azure-core-http-vertx/spotbugs-exclude.xml b/sdk/core/azure-core-http-vertx/spotbugs-exclude.xml index 034d19a0d9fa..9e5076b8694f 100644 --- a/sdk/core/azure-core-http-vertx/spotbugs-exclude.xml +++ b/sdk/core/azure-core-http-vertx/spotbugs-exclude.xml @@ -16,4 +16,12 @@ + + + + + + + + diff --git a/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClient.java b/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClient.java index 7083892be4d2..2ea83eb4d1cd 100644 --- a/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClient.java +++ b/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClient.java @@ -11,51 +11,55 @@ import com.azure.core.http.vertx.implementation.BufferedVertxHttpResponse; import com.azure.core.http.vertx.implementation.VertxHttpAsyncResponse; import com.azure.core.http.vertx.implementation.VertxRequestWriteSubscriber; +import com.azure.core.http.vertx.implementation.VertxUtils; import com.azure.core.implementation.util.BinaryDataContent; import com.azure.core.implementation.util.BinaryDataHelper; import com.azure.core.implementation.util.ByteArrayContent; import com.azure.core.implementation.util.ByteBufferContent; -import com.azure.core.implementation.util.FileContent; +import com.azure.core.implementation.util.HttpUtils; import com.azure.core.implementation.util.SerializableContent; import com.azure.core.implementation.util.StringContent; import com.azure.core.util.BinaryData; import com.azure.core.util.Context; import com.azure.core.util.Contexts; import com.azure.core.util.ProgressReporter; +import com.azure.core.util.logging.ClientLogger; import io.netty.buffer.Unpooled; -import io.vertx.core.Vertx; +import io.vertx.core.Future; +import io.vertx.core.Promise; import io.vertx.core.buffer.Buffer; -import io.vertx.core.file.AsyncFile; -import io.vertx.core.file.OpenOptions; import io.vertx.core.http.HttpClientRequest; import io.vertx.core.http.HttpClientResponse; import io.vertx.core.http.HttpMethod; import io.vertx.core.http.RequestOptions; import reactor.core.publisher.Mono; -import reactor.core.publisher.MonoSink; +import reactor.util.context.ContextView; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.time.Duration; import java.util.Objects; - -import static com.azure.core.http.vertx.implementation.VertxUtils.wrapVertxException; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; /** * {@link HttpClient} implementation for the Vert.x {@link io.vertx.core.http.HttpClient}. */ class VertxAsyncHttpClient implements HttpClient { - private static final String AZURE_EAGERLY_READ_RESPONSE = "azure-eagerly-read-response"; - private static final String AZURE_IGNORE_RESPONSE_BODY = "azure-ignore-response-body"; + private static final ClientLogger LOGGER = new ClientLogger(VertxAsyncHttpClient.class); - private final Vertx vertx; final io.vertx.core.http.HttpClient client; + private final Duration responseTimeout; /** * Constructs a {@link VertxAsyncHttpClient}. * * @param client The Vert.x {@link io.vertx.core.http.HttpClient} */ - VertxAsyncHttpClient(io.vertx.core.http.HttpClient client, Vertx vertx) { + VertxAsyncHttpClient(io.vertx.core.http.HttpClient client, Duration responseTimeout) { this.client = Objects.requireNonNull(client, "client cannot be null"); - this.vertx = Objects.requireNonNull(vertx, "vertx cannot be null"); + this.responseTimeout = responseTimeout; } @Override @@ -65,17 +69,57 @@ public Mono send(HttpRequest request) { @Override public Mono send(HttpRequest request, Context context) { - boolean eagerlyReadResponse = (boolean) context.getData(AZURE_EAGERLY_READ_RESPONSE).orElse(false); - boolean ignoreResponseBody = (boolean) context.getData(AZURE_IGNORE_RESPONSE_BODY).orElse(false); + return Mono.deferContextual(contextView -> Mono.fromFuture(sendInternal(request, context, contextView))) + .onErrorMap(VertxUtils::wrapVertxException); + } + + @Override + public HttpResponse sendSync(HttpRequest request, Context context) { + try { + return sendInternal(request, context, reactor.util.context.Context.empty()).get(); + } catch (Exception e) { + Throwable mapped = e; + if (e instanceof ExecutionException) { + mapped = e.getCause(); + } + + mapped = VertxUtils.wrapVertxException(mapped); + if (mapped instanceof Error) { + throw LOGGER.logThrowableAsError((Error) mapped); + } else if (mapped instanceof IOException) { + throw LOGGER.logExceptionAsError(new UncheckedIOException((IOException) mapped)); + } else if (mapped instanceof RuntimeException) { + throw LOGGER.logExceptionAsError((RuntimeException) mapped); + } else { + throw LOGGER.logExceptionAsError(new RuntimeException(mapped)); + } + } + } + + private CompletableFuture sendInternal(HttpRequest request, Context context, + ContextView contextView) { + boolean eagerlyReadResponse = (boolean) context.getData(HttpUtils.AZURE_EAGERLY_READ_RESPONSE).orElse(false); + boolean ignoreResponseBody = (boolean) context.getData(HttpUtils.AZURE_IGNORE_RESPONSE_BODY).orElse(false); + Duration perCallTimeout = (Duration) context.getData(HttpUtils.AZURE_RESPONSE_TIMEOUT) + .filter(timeoutDuration -> timeoutDuration instanceof Duration) + .orElse(responseTimeout); ProgressReporter progressReporter = Contexts.with(context).getHttpRequestProgressReporter(); RequestOptions options = new RequestOptions().setMethod(HttpMethod.valueOf(request.getHttpMethod().name())) .setAbsoluteURI(request.getUrl()); - return Mono.create(sink -> client.request(options, requestResult -> { + // This is the shared design for sending a request and receiving a response in Vert.x. The design relies on + // using a Vert.x Promise and the handler methods to control the flow of the request-response cycle. The Promise + // will be completed upon failure during the request-response cycle or upon completion of the response. This can + // be shared between sync and async as Reactor can use Mono.fromCompletionStage to convert the Promise to a Mono + // and the sync path can convert the Promise to a Future and block it. + + // Create the Promise that will be returned. Promise.promise() is an uncompleted Promise. + Promise promise = Promise.promise(); + client.request(options, requestResult -> { if (requestResult.failed()) { - sink.error(wrapVertxException(requestResult.cause())); + promise.fail(requestResult.cause()); return; } @@ -89,98 +133,70 @@ public Mono send(HttpRequest request, Context context) { vertxRequest.setChunked(true); } - vertxRequest.response(event -> { - if (event.succeeded()) { - HttpClientResponse vertxHttpResponse = event.result(); - vertxHttpResponse.exceptionHandler(exception -> sink.error(wrapVertxException(exception))); - - // TODO (alzimmer) - // For now Vertx will always use a buffered response until reliability issues when using streaming - // can be resolved. - if (eagerlyReadResponse || ignoreResponseBody) { - vertxHttpResponse.body(bodyEvent -> { - if (bodyEvent.succeeded()) { - sink.success( - new BufferedVertxHttpResponse(request, vertxHttpResponse, bodyEvent.result())); - } else { - sink.error(wrapVertxException(bodyEvent.cause())); - } - }); + Future responseFuture; + if (!perCallTimeout.isZero() && !perCallTimeout.isNegative()) { + responseFuture = vertxRequest.response().timeout(perCallTimeout.toMillis(), TimeUnit.MILLISECONDS); + } else { + responseFuture = vertxRequest.response(); + } + + responseFuture = responseFuture.onFailure(promise::fail); + + if (eagerlyReadResponse || ignoreResponseBody) { + responseFuture.andThen(responseResult -> { + if (responseResult.failed()) { + promise.fail(responseResult.cause()); + return; + } + + HttpClientResponse vertxHttpResponse = responseResult.result(); + vertxHttpResponse.body().andThen(bodyResult -> { + if (bodyResult.succeeded()) { + promise.complete( + new BufferedVertxHttpResponse(request, vertxHttpResponse, bodyResult.result())); + } else { + promise.fail(bodyResult.cause()); + } + }); + }); + } else { + responseFuture.andThen(responseResult -> { + if (responseResult.succeeded()) { + promise.complete(new VertxHttpAsyncResponse(request, responseResult.result())); } else { - sink.success(new VertxHttpAsyncResponse(request, vertxHttpResponse)); + promise.fail(responseResult.cause()); } - } else { - sink.error(wrapVertxException(event.cause())); - } - }); + }); + } - sendBody(sink, request, progressReporter, vertxRequest); - })); - } + sendBody(contextView, request, progressReporter, vertxRequest, promise); + }); - @Override - public HttpResponse sendSync(HttpRequest request, Context context) { - return send(request, context).block(); + return promise.future().toCompletionStage().toCompletableFuture(); } @SuppressWarnings("deprecation") - private void sendBody(MonoSink sink, HttpRequest azureRequest, ProgressReporter progressReporter, - HttpClientRequest vertxRequest) { + private void sendBody(ContextView contextView, HttpRequest azureRequest, ProgressReporter progressReporter, + HttpClientRequest vertxRequest, Promise promise) { BinaryData body = azureRequest.getBodyAsBinaryData(); - if (body == null) { - vertxRequest.send(result -> { - if (result.failed()) { - sink.error(wrapVertxException(result.cause())); - } - }); - return; - } - - BinaryDataContent bodyContent = BinaryDataHelper.getContent(body); - if (bodyContent instanceof ByteArrayContent - || bodyContent instanceof StringContent - || bodyContent instanceof SerializableContent) { - byte[] content = bodyContent.toBytes(); - vertxRequest.send(Buffer.buffer(Unpooled.wrappedBuffer(content)), result -> { - if (result.succeeded()) { - reportProgress(content.length, progressReporter); - } else { - sink.error(wrapVertxException(result.cause())); - } - }); - } else if (bodyContent instanceof ByteBufferContent) { - long contentLength = bodyContent.getLength(); - vertxRequest.send(Buffer.buffer(Unpooled.wrappedBuffer(bodyContent.toByteBuffer())), result -> { - if (result.succeeded()) { - reportProgress(contentLength, progressReporter); - } else { - sink.error(wrapVertxException(result.cause())); - } - }); - } else if (bodyContent instanceof FileContent) { - FileContent fileContent = (FileContent) bodyContent; - vertx.fileSystem().open(fileContent.getFile().toString(), new OpenOptions().setRead(true), event -> { - if (event.succeeded()) { - AsyncFile file = event.result(); - file.setReadPos(fileContent.getPosition()); - if (fileContent.getLength() != null) { - file.setReadLength(fileContent.getLength()); - } - vertxRequest.send(file, result -> { - if (result.succeeded()) { - reportProgress(fileContent.getLength(), progressReporter); - } else { - sink.error(wrapVertxException(result.cause())); - } - }); - } else { - sink.error(wrapVertxException(event.cause())); - } - }); + if (body == null) { + vertxRequest.send().onFailure(promise::fail); } else { - // Right now both Flux and InputStream bodies are being handled reactively. - azureRequest.getBody().subscribe(new VertxRequestWriteSubscriber(vertxRequest, sink, progressReporter)); + BinaryDataContent bodyContent = BinaryDataHelper.getContent(body); + if (bodyContent instanceof ByteArrayContent + || bodyContent instanceof ByteBufferContent + || bodyContent instanceof StringContent + || bodyContent instanceof SerializableContent) { + long contentLength = bodyContent.getLength(); + vertxRequest.send(Buffer.buffer(Unpooled.wrappedBuffer(bodyContent.toByteBuffer()))) + .onSuccess(ignored -> reportProgress(contentLength, progressReporter)) + .onFailure(promise::fail); + } else { + // Right now both Flux and InputStream bodies are being handled reactively. + azureRequest.getBody() + .subscribe(new VertxRequestWriteSubscriber(vertxRequest, promise, progressReporter, contextView)); + } } } diff --git a/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClientBuilder.java b/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClientBuilder.java index 270479074f73..ceca3b582e78 100644 --- a/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClientBuilder.java +++ b/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClientBuilder.java @@ -20,10 +20,11 @@ import java.util.concurrent.TimeUnit; import java.util.regex.Pattern; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_READ_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT; -import static com.azure.core.util.CoreUtils.getDefaultTimeoutFromEnvironment; +import static com.azure.core.implementation.util.HttpUtils.getDefaultConnectTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultReadTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultResponseTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultWriteTimeout; +import static com.azure.core.implementation.util.HttpUtils.getTimeout; /** * Builds a {@link VertxAsyncHttpClient}. @@ -34,24 +35,11 @@ public class VertxAsyncHttpClientBuilder { private static final Pattern NON_PROXY_HOSTS_SPLIT = Pattern.compile("(? + * The connection timeout begins once the request attempts to connect to the remote host and finishes once the + * connection is resolved. + *

+ * If {@code connectTimeout} is null either {@link Configuration#PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT} or a + * 10-second timeout will be used, if it is a {@link Duration} less than or equal to zero then no timeout will be + * applied. When applying the timeout the greatest of one millisecond and the value of {@code connectTimeout} will + * be used. *

- * The default read idle timeout is 60 seconds. + * By default, the connection timeout is 10 seconds. * - * @param readIdleTimeout the read idle timeout - * @return the updated VertxAsyncHttpClientBuilder object + * @param connectTimeout Connect timeout duration. + * @return The updated OkHttpAsyncHttpClientBuilder object. */ - public VertxAsyncHttpClientBuilder readIdleTimeout(Duration readIdleTimeout) { - this.readIdleTimeout = readIdleTimeout; + public VertxAsyncHttpClientBuilder connectTimeout(Duration connectTimeout) { + // setConnectionTimeout can be null + this.connectTimeout = connectTimeout; return this; } /** - * Sets the write idle timeout. + * Sets the read timeout duration used when reading the server response. *

- * The default read idle timeout is 60 seconds. + * The read timeout begins once the first response read is triggered after the server response is received. This + * timeout triggers periodically but won't fire its operation if another read operation has completed between when + * the timeout is triggered and completes. + *

+ * If {@code readTimeout} is null or {@link Configuration#PROPERTY_AZURE_REQUEST_READ_TIMEOUT} or a 60-second + * timeout will be used, if it is a {@link Duration} less than or equal to zero then no timeout period will be + * applied to response read. When applying the timeout the greatest of one millisecond and the value of {@code + * readTimeout} will be used. * - * @param writeIdleTimeout the write idle timeout - * @return the updated VertxAsyncHttpClientBuilder object + * @param readTimeout Read timeout duration. + * @return The updated OkHttpAsyncHttpClientBuilder object. */ - public VertxAsyncHttpClientBuilder writeIdleTimeout(Duration writeIdleTimeout) { - this.writeIdleTimeout = writeIdleTimeout; + public VertxAsyncHttpClientBuilder readTimeout(Duration readTimeout) { + this.readTimeout = readTimeout; return this; } /** - * Sets the connect timeout. + * Sets the response timeout duration used when waiting for a server to reply. + *

+ * The response timeout begins once the request write completes and finishes once the first response read is + * triggered when the server response is received. + *

+ * If {@code responseTimeout} is null either {@link Configuration#PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT} or a + * 60-second timeout will be used, if it is a {@link Duration} less than or equal to zero then no timeout will be + * applied to the response. When applying the timeout the greatest of one millisecond and the value of {@code + * responseTimeout} will be used. *

- * The default connect timeout is 10 seconds. + * Given OkHttp doesn't have an equivalent timeout for just responses, this is handled manually. * - * @param connectTimeout the connection timeout - * @return the updated VertxAsyncHttpClientBuilder object + * @param responseTimeout Response timeout duration. + * @return The updated VertxAsyncHttpClientBuilder object. */ - public VertxAsyncHttpClientBuilder connectTimeout(Duration connectTimeout) { - this.connectTimeout = connectTimeout; + public VertxAsyncHttpClientBuilder responseTimeout(Duration responseTimeout) { + this.responseTimeout = responseTimeout; return this; } /** - * Sets the connection idle timeout. + * Sets the writing timeout for a request to be sent. + *

+ * The writing timeout does not apply to the entire request but to the request being sent over the wire. For example + * a request body which emits {@code 10} {@code 8KB} buffers will trigger {@code 10} write operations, the last + * write tracker will update when each operation completes and the outbound buffer will be periodically checked to + * determine if it is still draining. *

- * The default connect timeout is 60 seconds. + * If {@code writeTimeout} is null either {@link Configuration#PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT} or a 60-second + * timeout will be used, if it is a {@link Duration} less than or equal to zero then no write timeout will be + * applied. When applying the timeout the greatest of one millisecond and the value of {@code writeTimeout} will be + * used. * - * @param idleTimeout the connection idle timeout - * @return the updated VertxAsyncHttpClientBuilder object + * @param writeTimeout Write operation timeout duration. + * @return The updated VertxAsyncHttpClientBuilder object. */ - public VertxAsyncHttpClientBuilder idleTimeout(Duration idleTimeout) { - this.idleTimeout = idleTimeout; + public VertxAsyncHttpClientBuilder writeTimeout(Duration writeTimeout) { + this.writeTimeout = writeTimeout; return this; } @@ -178,35 +198,18 @@ public HttpClient build() { configuredVertx = getVertx(vertxProviders.iterator()); } - if (this.httpClientOptions == null) { - this.httpClientOptions = new HttpClientOptions(); - - if (this.connectTimeout != null) { - this.httpClientOptions.setConnectTimeout((int) this.connectTimeout.toMillis()); - } else { - this.httpClientOptions.setConnectTimeout((int) DEFAULT_CONNECT_TIMEOUT); - } - - if (this.readIdleTimeout != null) { - this.httpClientOptions.setReadIdleTimeout((int) this.readIdleTimeout.getSeconds()); - } else { - this.httpClientOptions.setReadIdleTimeout((int) DEFAULT_READ_TIMEOUT); - } - - if (this.writeIdleTimeout != null) { - this.httpClientOptions.setWriteIdleTimeout((int) this.writeIdleTimeout.getSeconds()); - } else { - this.httpClientOptions.setWriteIdleTimeout((int) DEFAULT_WRITE_TIMEOUT); - } - - this.httpClientOptions.setIdleTimeout((int) this.idleTimeout.getSeconds()); + HttpClientOptions buildOptions = this.httpClientOptions; + if (buildOptions == null) { + buildOptions = new HttpClientOptions().setIdleTimeoutUnit(TimeUnit.MILLISECONDS) + .setConnectTimeout((int) getTimeout(this.connectTimeout, getDefaultConnectTimeout()).toMillis()) + .setReadIdleTimeout((int) getTimeout(this.readTimeout, getDefaultReadTimeout()).toMillis()) + .setWriteIdleTimeout((int) getTimeout(this.writeTimeout, getDefaultWriteTimeout()).toMillis()); Configuration buildConfiguration - = (this.configuration == null) ? Configuration.getGlobalConfiguration() : configuration; + = (configuration == null) ? Configuration.getGlobalConfiguration() : configuration; - ProxyOptions buildProxyOptions = (this.proxyOptions == null) - ? ProxyOptions.fromConfiguration(buildConfiguration, true) - : this.proxyOptions; + ProxyOptions buildProxyOptions + = (proxyOptions == null) ? ProxyOptions.fromConfiguration(buildConfiguration, true) : proxyOptions; if (buildProxyOptions != null) { io.vertx.core.net.ProxyOptions vertxProxyOptions = new io.vertx.core.net.ProxyOptions(); @@ -238,16 +241,16 @@ public HttpClient build() { String nonProxyHosts = buildProxyOptions.getNonProxyHosts(); if (!CoreUtils.isNullOrEmpty(nonProxyHosts)) { for (String nonProxyHost : desanitizedNonProxyHosts(nonProxyHosts)) { - this.httpClientOptions.addNonProxyHost(nonProxyHost); + buildOptions.addNonProxyHost(nonProxyHost); } } - this.httpClientOptions.setProxyOptions(vertxProxyOptions); + buildOptions.setProxyOptions(vertxProxyOptions); } } - io.vertx.core.http.HttpClient client = configuredVertx.createHttpClient(this.httpClientOptions); - return new VertxAsyncHttpClient(client, configuredVertx); + io.vertx.core.http.HttpClient client = configuredVertx.createHttpClient(buildOptions); + return new VertxAsyncHttpClient(client, getTimeout(this.responseTimeout, getDefaultResponseTimeout())); } static Vertx getVertx(Iterator iterator) { diff --git a/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClientProvider.java b/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClientProvider.java index 46c1a2476340..fba579502404 100644 --- a/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClientProvider.java +++ b/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxAsyncHttpClientProvider.java @@ -60,9 +60,9 @@ public HttpClient createInstance(HttpClientOptions clientOptions) { return new VertxAsyncHttpClientBuilder().proxy(clientOptions.getProxyOptions()) .configuration(clientOptions.getConfiguration()) .connectTimeout(clientOptions.getConnectTimeout()) - .idleTimeout(clientOptions.getConnectionIdleTimeout()) - .writeIdleTimeout(clientOptions.getWriteTimeout()) - .readIdleTimeout(clientOptions.getReadTimeout()) + .writeTimeout(clientOptions.getWriteTimeout()) + .responseTimeout(clientOptions.getResponseTimeout()) + .readTimeout(clientOptions.getReadTimeout()) .build(); } } diff --git a/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/implementation/VertxRequestWriteSubscriber.java b/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/implementation/VertxRequestWriteSubscriber.java index 7324a2bf38a0..2c9d12e16870 100644 --- a/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/implementation/VertxRequestWriteSubscriber.java +++ b/sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/implementation/VertxRequestWriteSubscriber.java @@ -5,7 +5,6 @@ import com.azure.core.http.HttpResponse; import com.azure.core.util.ProgressReporter; -import com.azure.core.util.logging.ClientLogger; import io.netty.buffer.Unpooled; import io.vertx.core.buffer.Buffer; import io.vertx.core.http.HttpClientRequest; @@ -14,21 +13,19 @@ import reactor.core.publisher.MonoSink; import reactor.core.publisher.Operators; import reactor.util.context.Context; +import reactor.util.context.ContextView; import java.nio.ByteBuffer; -import static com.azure.core.http.vertx.implementation.VertxUtils.wrapVertxException; - /** * Subscriber that writes a stream of {@link ByteBuffer ByteBuffers} to a {@link HttpClientRequest Vert.x request}. */ @SuppressWarnings("ReactiveStreamsSubscriberImplementation") public final class VertxRequestWriteSubscriber implements Subscriber { - private static final ClientLogger LOGGER = new ClientLogger(VertxRequestWriteSubscriber.class); - private final HttpClientRequest request; - private final MonoSink emitter; + private final io.vertx.core.Promise promise; private final ProgressReporter progressReporter; + private final ContextView contextView; // This subscriber is effectively synchronous so there is no need for these fields to be volatile. private volatile Subscription subscription; @@ -41,14 +38,16 @@ public final class VertxRequestWriteSubscriber implements Subscriber * {@link HttpClientRequest Vert.x request}. * * @param request The {@link HttpClientRequest Vert.x request} to write to. - * @param emitter The {@link MonoSink} to emit the {@link HttpResponse response} to. + * @param promise The {@link MonoSink} to emit the {@link HttpResponse response} to. * @param progressReporter The {@link ProgressReporter} to report progress to. + * @param contextView The {@link ContextView} to use when dropping errors. */ - public VertxRequestWriteSubscriber(HttpClientRequest request, MonoSink emitter, - ProgressReporter progressReporter) { + public VertxRequestWriteSubscriber(HttpClientRequest request, io.vertx.core.Promise promise, + ProgressReporter progressReporter, ContextView contextView) { this.request = request.exceptionHandler(this::onError).drainHandler(ignored -> requestNext()); - this.emitter = emitter; + this.promise = promise; this.progressReporter = progressReporter; + this.contextView = contextView; } @Override @@ -122,7 +121,7 @@ private void onErrorInternal(Throwable throwable) { State state = this.state; // code 2 and greater are completion states which means the error should be dropped as we already completed. if (state.code >= 2) { - Operators.onErrorDropped(throwable, Context.of(emitter.contextView())); + Operators.onErrorDropped(throwable, Context.of(contextView)); } this.state = State.ERROR; @@ -135,7 +134,7 @@ private void onErrorInternal(Throwable throwable) { private void resetRequest(Throwable throwable) { subscription.cancel(); - emitter.error(wrapVertxException(throwable)); + promise.fail(throwable); request.reset(0, throwable); } @@ -156,11 +155,7 @@ public void onComplete() { } private void endRequest() { - request.end(result -> { - if (result.failed()) { - emitter.error(wrapVertxException(result.cause())); - } - }); + request.end().onFailure(promise::fail); } private enum State { diff --git a/sdk/core/azure-core-http-vertx/src/samples/java/com/azure/core/http/vertx/VertxHttpClientBuilderJavaDocCodeSnippets.java b/sdk/core/azure-core-http-vertx/src/samples/java/com/azure/core/http/vertx/VertxHttpClientBuilderJavaDocCodeSnippets.java index 8a8da69d7664..6150813bb45d 100644 --- a/sdk/core/azure-core-http-vertx/src/samples/java/com/azure/core/http/vertx/VertxHttpClientBuilderJavaDocCodeSnippets.java +++ b/sdk/core/azure-core-http-vertx/src/samples/java/com/azure/core/http/vertx/VertxHttpClientBuilderJavaDocCodeSnippets.java @@ -12,6 +12,7 @@ /** * Code snippets for {@link VertxHttpClientBuilderJavaDocCodeSnippets} */ +@SuppressWarnings("unused") public class VertxHttpClientBuilderJavaDocCodeSnippets { /** @@ -55,25 +56,15 @@ public void proxyBasicAuthenticationSample() { } - public void connectionTimeoutSample() { - - // BEGIN: com.azure.core.http.vertx.vertxAsyncHttpClientBuilder#connextTimeout - final Duration connextTimeout = Duration.ofSeconds(250); // connection timeout of 250 seconds - HttpClient client = new VertxAsyncHttpClientBuilder() - .connectTimeout(connextTimeout) - .build(); - // END: com.azure.core.http.vertx.vertxAsyncHttpClientBuilder#connextTimeout - - } - - public void readTimeoutSample() { - - // BEGIN: com.azure.core.http.vertx.vertxAsyncHttpClientBuilder#readTimeout - final Duration readIdleTimeout = Duration.ofSeconds(100); // read timeout of 100 seconds + public void timeoutSample() { + // BEGIN: com.azure.core.http.vertx.VertxAsyncHttpClientBuilder#timeoutSample HttpClient client = new VertxAsyncHttpClientBuilder() - .readIdleTimeout(readIdleTimeout) - .build(); - // END: com.azure.core.http.vertx.vertxAsyncHttpClientBuilder#readTimeout + .connectTimeout(Duration.ofSeconds(10)) // Timeout of 10 seconds for establishing a connection + .writeTimeout(Duration.ofSeconds(100)) // Timeout of 100 seconds when network writing is idle + .responseTimeout(Duration.ofSeconds(30)) // Timeout of 30 seconds for the server to return a response + .readTimeout(Duration.ofSeconds(100)) // Timeout of 100 seconds when network reading is idle + .build(); + // END: com.azure.core.http.vertx.VertxAsyncHttpClientBuilder#timeoutSample } } diff --git a/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientBuilderTests.java b/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientBuilderTests.java index 56a9bade3cb1..6a642cbf9ee5 100644 --- a/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientBuilderTests.java +++ b/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientBuilderTests.java @@ -63,8 +63,7 @@ public void buildWithConfigurationNone() { @Test public void buildWithDefaultConnectionOptions() { - VertxAsyncHttpClientBuilder builder = new VertxAsyncHttpClientBuilder(); - HttpClient httpClient = builder.build(); + HttpClient httpClient = new VertxAsyncHttpClientBuilder().build(); io.vertx.core.http.HttpClient client = ((VertxAsyncHttpClient) httpClient).client; io.vertx.core.http.HttpClientOptions options = ((HttpClientImpl) client).options(); @@ -76,18 +75,16 @@ public void buildWithDefaultConnectionOptions() { .verifyComplete(); assertEquals(10000, options.getConnectTimeout()); - assertEquals(60, options.getIdleTimeout()); - assertEquals(60, options.getReadIdleTimeout()); - assertEquals(60, options.getWriteIdleTimeout()); + assertEquals(60000, options.getReadIdleTimeout()); + assertEquals(60000, options.getWriteIdleTimeout()); } @Test public void buildWithConnectionOptions() { VertxAsyncHttpClientBuilder builder = new VertxAsyncHttpClientBuilder(); VertxAsyncHttpClient httpClient = (VertxAsyncHttpClient) builder.connectTimeout(Duration.ofSeconds(10)) - .idleTimeout(Duration.ofSeconds(20)) - .readIdleTimeout(Duration.ofSeconds(30)) - .writeIdleTimeout(Duration.ofSeconds(40)) + .readTimeout(Duration.ofSeconds(30)) + .writeTimeout(Duration.ofSeconds(40)) .build(); io.vertx.core.http.HttpClientOptions options = ((HttpClientImpl) httpClient.client).options(); @@ -99,9 +96,8 @@ public void buildWithConnectionOptions() { .verifyComplete(); assertEquals(10000, options.getConnectTimeout()); - assertEquals(20, options.getIdleTimeout()); - assertEquals(30, options.getReadIdleTimeout()); - assertEquals(40, options.getWriteIdleTimeout()); + assertEquals(30000, options.getReadIdleTimeout()); + assertEquals(40000, options.getWriteIdleTimeout()); } @ParameterizedTest @@ -214,15 +210,13 @@ public void buildWithCustomHttpClientOptions() { options.setWriteIdleTimeout(70); HttpClient httpClient = new VertxAsyncHttpClientBuilder().connectTimeout(Duration.ofSeconds(10)) - .idleTimeout(Duration.ofSeconds(20)) - .readIdleTimeout(Duration.ofSeconds(30)) - .writeIdleTimeout(Duration.ofSeconds(40)) + .readTimeout(Duration.ofSeconds(30)) + .writeTimeout(Duration.ofSeconds(40)) .httpClientOptions(options) .build(); // Verify the original configuration was preserved and not overwritten assertEquals(30000, options.getConnectTimeout()); - assertEquals(50, options.getIdleTimeout()); assertEquals(60, options.getReadIdleTimeout()); assertEquals(70, options.getWriteIdleTimeout()); diff --git a/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientProviderTests.java b/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientProviderTests.java index e7e68f08c110..1714f4a0f566 100644 --- a/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientProviderTests.java +++ b/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientProviderTests.java @@ -107,9 +107,8 @@ public void optionsWithTimeouts() { io.vertx.core.http.HttpClientOptions options = ((HttpClientImpl) httpClient.client).options(); assertEquals(timeout.toMillis(), options.getConnectTimeout()); - assertEquals(timeout.getSeconds(), options.getIdleTimeout()); - assertEquals(timeout.getSeconds(), options.getReadIdleTimeout()); - assertEquals(timeout.getSeconds(), options.getWriteIdleTimeout()); + assertEquals(timeout.toMillis(), options.getReadIdleTimeout()); + assertEquals(timeout.toMillis(), options.getWriteIdleTimeout()); } @Test diff --git a/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientTests.java b/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientTests.java index 8e3e699f49f4..fa4742430177 100644 --- a/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientTests.java +++ b/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxAsyncHttpClientTests.java @@ -10,10 +10,10 @@ import com.azure.core.http.HttpMethod; import com.azure.core.http.HttpRequest; import com.azure.core.http.HttpResponse; +import com.azure.core.implementation.util.HttpUtils; import com.azure.core.util.Context; +import com.azure.core.util.FluxUtil; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.parallel.Execution; -import org.junit.jupiter.api.parallel.ExecutionMode; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.publisher.ParallelFlux; @@ -36,16 +36,20 @@ import java.util.concurrent.Callable; import java.util.concurrent.ForkJoinPool; import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import static com.azure.core.http.vertx.VertxHttpClientLocalTestServer.LONG_BODY; import static com.azure.core.http.vertx.VertxHttpClientLocalTestServer.RETURN_HEADERS_AS_IS_PATH; import static com.azure.core.http.vertx.VertxHttpClientLocalTestServer.SHORT_BODY; +import static com.azure.core.http.vertx.VertxHttpClientLocalTestServer.TIMEOUT; +import static com.azure.core.test.utils.TestUtils.assertArraysEqual; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.junit.jupiter.api.Assertions.assertLinesMatch; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; -@Execution(ExecutionMode.SAME_THREAD) public class VertxAsyncHttpClientTests { private static final String SERVER_HTTP_URI = VertxHttpClientLocalTestServer.getServer().getHttpUri(); @@ -236,6 +240,46 @@ public void testEmptyBufferedResponse() { .verifyComplete(); } + @Test + public void perCallTimeout() { + HttpClient client = new VertxAsyncHttpClientBuilder().responseTimeout(Duration.ofSeconds(10)).build(); + + HttpRequest request = new HttpRequest(HttpMethod.GET, url(TIMEOUT)); + + // Verify a smaller timeout sent through Context times out the request. + StepVerifier.create(client.send(request, new Context(HttpUtils.AZURE_RESPONSE_TIMEOUT, Duration.ofSeconds(1)))) + .expectErrorMatches(e -> e instanceof TimeoutException) + .verify(); + + // Then verify not setting a timeout through Context does not time out the request. + StepVerifier.create(client.send(request) + .flatMap(response -> Mono.zip(FluxUtil.collectBytesInByteBufferStream(response.getBody()), + Mono.just(response.getStatusCode())))) + .assertNext(tuple -> { + assertArraysEqual(SHORT_BODY, tuple.getT1()); + assertEquals(200, tuple.getT2()); + }) + .verifyComplete(); + } + + @Test + public void perCallTimeoutSync() { + HttpClient client = new VertxAsyncHttpClientBuilder().responseTimeout(Duration.ofSeconds(10)).build(); + + HttpRequest request = new HttpRequest(HttpMethod.GET, url(TIMEOUT)); + + // Verify a smaller timeout sent through Context times out the request. + RuntimeException ex = assertThrows(RuntimeException.class, + () -> client.sendSync(request, new Context(HttpUtils.AZURE_RESPONSE_TIMEOUT, Duration.ofSeconds(1)))); + assertInstanceOf(TimeoutException.class, ex.getCause()); + + // Then verify not setting a timeout through Context does not time out the request. + try (HttpResponse response = client.sendSync(request, Context.NONE)) { + assertEquals(200, response.getStatusCode()); + assertArraysEqual(SHORT_BODY, response.getBodyAsByteArray().block()); + } + } + private static Mono getResponse(String path) { HttpClient client = new VertxAsyncHttpClientBuilder().build(); return getResponse(client, path, Context.NONE); diff --git a/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxHttpClientLocalTestServer.java b/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxHttpClientLocalTestServer.java index 01c63e032632..1d533180191a 100644 --- a/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxHttpClientLocalTestServer.java +++ b/sdk/core/azure-core-http-vertx/src/test/java/com/azure/core/http/vertx/VertxHttpClientLocalTestServer.java @@ -30,6 +30,7 @@ public final class VertxHttpClientLocalTestServer { public static final byte[] SHORT_BODY = "hi there".getBytes(StandardCharsets.UTF_8); public static final byte[] LONG_BODY = createLongBody(); public static final String RETURN_HEADERS_AS_IS_PATH = "/returnHeadersAsIs"; + public static final String TIMEOUT = "/timeout"; public static final String PROXY_USERNAME = "foo"; public static final String PROXY_PASSWORD = "bar"; @@ -109,6 +110,16 @@ private static LocalTestServer initializeServer() { resp.setContentLength(0); } else if (get && "/connectionClose".equals(path)) { resp.getHttpChannel().getConnection().close(); + } else if (get && TIMEOUT.equals(path)) { + try { + Thread.sleep(5000); + resp.setStatus(200); + resp.getHttpOutput().write(SHORT_BODY); + resp.getHttpOutput().flush(); + resp.getHttpOutput().complete(Callback.NOOP); + } catch (InterruptedException e) { + throw new ServletException(e); + } } else { throw new ServletException("Unexpected request " + req.getMethod() + " " + path); } diff --git a/sdk/core/azure-core-test/src/main/java/com/azure/core/test/http/HttpClientTests.java b/sdk/core/azure-core-test/src/main/java/com/azure/core/test/http/HttpClientTests.java index cd1a10576a7d..ae9d33241b2e 100644 --- a/sdk/core/azure-core-test/src/main/java/com/azure/core/test/http/HttpClientTests.java +++ b/sdk/core/azure-core-test/src/main/java/com/azure/core/test/http/HttpClientTests.java @@ -584,7 +584,7 @@ private static void canSendBinaryDataSync(URL requestUrl, BinaryData requestBody HttpRequest request = new HttpRequest(HttpMethod.PUT, requestUrl, new HttpHeaders(), requestBody); try (HttpResponse httpResponse = httpClient.sendSync(request, Context.NONE)) { - byte[] responseBytes = httpResponse.getBodyAsByteArray().block(); + byte[] responseBytes = httpResponse.getBodyAsBinaryData().toBytes(); assertArraysEqual(expectedResponseBody, responseBytes); } } @@ -615,7 +615,7 @@ private static void canSendBinaryDataSyncWithProgressReporter(URL requestUrl, Bi .getContext(); try (HttpResponse httpResponse = httpClient.sendSync(request, context)) { - byte[] responseBytes = httpResponse.getBodyAsByteArray().block(); + byte[] responseBytes = httpResponse.getBodyAsBinaryData().toBytes(); assertArraysEqual(expectedResponseBody, responseBytes); assertEquals(expectedResponseBody.length, progress.intValue()); } diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/rest/RestProxyBase.java b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/rest/RestProxyBase.java index 67f4ea40662e..a33abec128a8 100644 --- a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/rest/RestProxyBase.java +++ b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/http/rest/RestProxyBase.java @@ -27,6 +27,7 @@ import com.azure.core.implementation.http.UnexpectedExceptionInformation; import com.azure.core.implementation.serializer.HttpResponseDecoder; import com.azure.core.implementation.serializer.MalformedValueException; +import com.azure.core.implementation.util.HttpUtils; import com.azure.core.util.BinaryData; import com.azure.core.util.Context; import com.azure.core.util.UrlBuilder; @@ -109,15 +110,15 @@ public final Object invoke(Object proxy, Method method, RequestOptions options, // For the following Context options only set a value if it's true. This is to avoid adding a key to the // context with a value of false, which will increase all subsequent lookups of the context. if (methodParser.isResponseEagerlyRead()) { - context = context.addData("azure-eagerly-read-response", true); + context = context.addData(HttpUtils.AZURE_EAGERLY_READ_RESPONSE, true); } if (methodParser.isResponseBodyIgnored()) { - context = context.addData("azure-ignore-response-body", true); + context = context.addData(HttpUtils.AZURE_IGNORE_RESPONSE_BODY, true); } if (methodParser.isHeadersEagerlyConverted()) { - context = context.addData("azure-eagerly-convert-headers", true); + context = context.addData(HttpUtils.AZURE_EAGERLY_CONVERT_HEADERS, true); } return invoke(proxy, method, options, errorOptions, requestCallback, methodParser, request, context); diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/HttpUtils.java b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/HttpUtils.java new file mode 100644 index 000000000000..0fd02b8f9f3a --- /dev/null +++ b/sdk/core/azure-core/src/main/java/com/azure/core/implementation/util/HttpUtils.java @@ -0,0 +1,130 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.azure.core.implementation.util; + +import com.azure.core.util.Configuration; +import com.azure.core.util.logging.ClientLogger; + +import java.time.Duration; + +import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT; +import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_READ_TIMEOUT; +import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT; +import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT; +import static com.azure.core.util.CoreUtils.getDefaultTimeoutFromEnvironment; + +/** + * Utilities shared with HttpClient implementations. + */ +public final class HttpUtils { + private static final ClientLogger LOGGER = new ClientLogger(HttpUtils.class); + + private static final Duration MINIMUM_TIMEOUT = Duration.ofMillis(1); + private static final Duration DEFAULT_CONNECT_TIMEOUT; + private static final Duration DEFAULT_WRITE_TIMEOUT; + private static final Duration DEFAULT_RESPONSE_TIMEOUT; + private static final Duration DEFAULT_READ_TIMEOUT; + + static { + Configuration configuration = Configuration.getGlobalConfiguration(); + DEFAULT_CONNECT_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, + PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT, Duration.ofSeconds(10), LOGGER); + DEFAULT_WRITE_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT, + Duration.ofSeconds(60), LOGGER); + DEFAULT_RESPONSE_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, + PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT, Duration.ofSeconds(60), LOGGER); + DEFAULT_READ_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, PROPERTY_AZURE_REQUEST_READ_TIMEOUT, + Duration.ofSeconds(60), LOGGER); + } + + /** + * Context key used to indicate to an HttpClient implementation if it should eagerly read the response from the + * network. + */ + public static final String AZURE_EAGERLY_READ_RESPONSE = "azure-eagerly-read-response"; + + /** + * Context key used to indicate to an HttpClient implementation if the response body should be ignored and eagerly + * drained from the network. + */ + public static final String AZURE_IGNORE_RESPONSE_BODY = "azure-ignore-response-body"; + + /** + * Context key used to indicate to an HttpClient a per-call response timeout. + */ + public static final String AZURE_RESPONSE_TIMEOUT = "azure-response-timeout"; + + /** + * Context key used to indicate to an HttpClient if the implementation specific HTTP headers should be converted to + * Azure Core HttpHeaders. + */ + public static final String AZURE_EAGERLY_CONVERT_HEADERS = "azure-eagerly-convert-headers"; + + /** + * Gets the default connect timeout. + * + * @return The default connect timeout. + */ + public static Duration getDefaultConnectTimeout() { + return DEFAULT_CONNECT_TIMEOUT; + } + + /** + * Gets the default write timeout. + * + * @return The default write timeout. + */ + public static Duration getDefaultWriteTimeout() { + return DEFAULT_WRITE_TIMEOUT; + } + + /** + * Gets the default response timeout. + * + * @return The default response timeout. + */ + public static Duration getDefaultResponseTimeout() { + return DEFAULT_RESPONSE_TIMEOUT; + } + + /** + * Gets the default read timeout. + * + * @return The default read timeout. + */ + public static Duration getDefaultReadTimeout() { + return DEFAULT_READ_TIMEOUT; + } + + /** + * Returns the timeout Duration to use based on the configured timeout and the default timeout. + *

+ * If the configured timeout is null the default timeout will be used. If the timeout is less than or equal to zero + * no timeout will be used. If the timeout is less than one millisecond a timeout of one millisecond will be used. + * + * @param configuredTimeout The configured timeout. + * @param defaultTimeout The default timeout. + * @return The timeout to use. + */ + public static Duration getTimeout(Duration configuredTimeout, Duration defaultTimeout) { + // Timeout is null, use the default timeout. + if (configuredTimeout == null) { + return defaultTimeout; + } + + // Timeout is less than or equal to zero, return no timeout. + if (configuredTimeout.isZero() || configuredTimeout.isNegative()) { + return Duration.ZERO; + } + + // Return the maximum of the timeout period and the minimum allowed timeout period. + if (configuredTimeout.compareTo(MINIMUM_TIMEOUT) < 0) { + return MINIMUM_TIMEOUT; + } else { + return configuredTimeout; + } + } + + private HttpUtils() { + } +} diff --git a/sdk/core/azure-core/src/main/java/com/azure/core/util/HttpClientOptions.java b/sdk/core/azure-core/src/main/java/com/azure/core/util/HttpClientOptions.java index 43808efe0eea..6d9a4e1bb2b6 100644 --- a/sdk/core/azure-core/src/main/java/com/azure/core/util/HttpClientOptions.java +++ b/sdk/core/azure-core/src/main/java/com/azure/core/util/HttpClientOptions.java @@ -11,11 +11,11 @@ import java.time.Duration; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_READ_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT; -import static com.azure.core.util.Configuration.PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT; -import static com.azure.core.util.CoreUtils.getDefaultTimeoutFromEnvironment; +import static com.azure.core.implementation.util.HttpUtils.getDefaultConnectTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultReadTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultResponseTimeout; +import static com.azure.core.implementation.util.HttpUtils.getDefaultWriteTimeout; +import static com.azure.core.implementation.util.HttpUtils.getTimeout; /** * General configuration options for {@link HttpClient HttpClients}. @@ -24,29 +24,10 @@ */ @Fluent public final class HttpClientOptions extends ClientOptions { - private static final Duration MINIMUM_TIMEOUT = Duration.ofMillis(1); - private static final Duration DEFAULT_CONNECT_TIMEOUT; - private static final Duration DEFAULT_WRITE_TIMEOUT; - private static final Duration DEFAULT_RESPONSE_TIMEOUT; - private static final Duration DEFAULT_READ_TIMEOUT; private static final Duration DEFAULT_CONNECTION_IDLE_TIMEOUT = Duration.ofSeconds(60); - private static final Duration NO_TIMEOUT = Duration.ZERO; private static final ClientLogger LOGGER = new ClientLogger(HttpClientOptions.class); - static { - Configuration configuration = Configuration.getGlobalConfiguration(); - - DEFAULT_CONNECT_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, - PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT, Duration.ofSeconds(10), LOGGER); - DEFAULT_WRITE_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT, - Duration.ofSeconds(60), LOGGER); - DEFAULT_RESPONSE_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, - PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT, Duration.ofSeconds(60), LOGGER); - DEFAULT_READ_TIMEOUT = getDefaultTimeoutFromEnvironment(configuration, PROPERTY_AZURE_REQUEST_READ_TIMEOUT, - Duration.ofSeconds(60), LOGGER); - } - private ProxyOptions proxyOptions; private Configuration configuration; private Duration connectTimeout; @@ -141,12 +122,20 @@ public HttpClientOptions setConnectTimeout(Duration connectTimeout) { /** * Gets the connection timeout for a request to be sent. *

+ * The connection timeout begins once the request attempts to connect to the remote host and finishes when the + * connection is resolved. + *

+ * If {@code connectTimeout} is null either {@link Configuration#PROPERTY_AZURE_REQUEST_CONNECT_TIMEOUT} or a + * 10-second timeout will be used, if it is a {@link Duration} less than or equal to zero then no timeout will be + * applied. When applying the timeout the greatest of one millisecond and the value of {@code connectTimeout} will + * be used. + *

* The default connection timeout is 10 seconds. * * @return The connection timeout of a request to be sent. */ public Duration getConnectTimeout() { - return getTimeout(connectTimeout, DEFAULT_CONNECT_TIMEOUT); + return getTimeout(connectTimeout, getDefaultConnectTimeout()); } /** @@ -174,12 +163,21 @@ public HttpClientOptions setWriteTimeout(Duration writeTimeout) { /** * Gets the writing timeout for a request to be sent. *

+ * The writing timeout does not apply to the entire request but to each emission being sent over the wire. For + * example a request body which emits {@code 10} {@code 8KB} buffers will trigger {@code 10} write operations, the + * outbound buffer will be periodically checked to determine if it is still draining. + *

+ * If {@code writeTimeout} is null either {@link Configuration#PROPERTY_AZURE_REQUEST_WRITE_TIMEOUT} or a 60-second + * timeout will be used, if it is a {@link Duration} less than or equal to zero then no write timeout will be + * applied. When applying the timeout the greatest of one millisecond and the value of {@code writeTimeout} will be + * used. + *

* The default writing timeout is 60 seconds. * * @return The writing timeout of a request to be sent. */ public Duration getWriteTimeout() { - return getTimeout(writeTimeout, DEFAULT_WRITE_TIMEOUT); + return getTimeout(writeTimeout, getDefaultWriteTimeout()); } /** @@ -227,12 +225,20 @@ public HttpClientOptions setResponseTimeout(Duration responseTimeout) { /** * Gets the response timeout duration used when waiting for a server to reply. *

+ * The response timeout begins once the request write completes and finishes once the first response read is + * triggered when the server response is received. + *

+ * If {@code responseTimeout} is null either {@link Configuration#PROPERTY_AZURE_REQUEST_RESPONSE_TIMEOUT} or a + * 60-second timeout will be used, if it is a {@link Duration} less than or equal to zero then no timeout will be + * applied to the response. When applying the timeout the greatest of one millisecond and the value of + * {@code responseTimeout} will be used. + *

* The default response timeout is 60 seconds. * * @return The response timeout duration. */ public Duration getResponseTimeout() { - return getTimeout(responseTimeout, DEFAULT_RESPONSE_TIMEOUT); + return getTimeout(responseTimeout, getDefaultResponseTimeout()); } /** @@ -287,7 +293,7 @@ public HttpClientOptions setReadTimeout(Duration readTimeout) { * @return The read timeout duration. */ public Duration getReadTimeout() { - return getTimeout(readTimeout, DEFAULT_READ_TIMEOUT); + return getTimeout(readTimeout, getDefaultReadTimeout()); } /** @@ -398,19 +404,4 @@ public HttpClientOptions setHttpClientProvider(Class getHttpClientProvider() { return httpClientProvider; } - - private static Duration getTimeout(Duration configuredTimeout, Duration defaultTimeout) { - // Timeout is null, use the default timeout. - if (configuredTimeout == null) { - return defaultTimeout; - } - - // Timeout is less than or equal to zero, return no timeout. - if (configuredTimeout.isZero() || configuredTimeout.isNegative()) { - return NO_TIMEOUT; - } - - // Return the maximum of the timeout period and the minimum allowed timeout period. - return configuredTimeout.compareTo(MINIMUM_TIMEOUT) > 0 ? configuredTimeout : MINIMUM_TIMEOUT; - } }