diff --git a/sdk/core/azure-core-http-netty/CHANGELOG.md b/sdk/core/azure-core-http-netty/CHANGELOG.md index db11cd9233df7..49f9a783fbaf7 100644 --- a/sdk/core/azure-core-http-netty/CHANGELOG.md +++ b/sdk/core/azure-core-http-netty/CHANGELOG.md @@ -3,6 +3,9 @@ ## 1.17.0-beta.1 (Unreleased) ### Features Added +- Added opt-in observation of the `Expect: 100-continue` handshake. When a caller places an `AtomicBoolean` in the + request `Context` under the key `azure-http-client-expect-continue-received`, the client sets it to `true` if the + service answers with an interim `100 Continue` response. This does not change how or when the request body is sent. ### Breaking Changes 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 a6a085fd40586..1da95d2e41faf 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 @@ -52,6 +52,7 @@ import java.nio.file.StandardOpenOption; import java.time.Duration; import java.util.Objects; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiFunction; /** @@ -95,6 +96,12 @@ class NettyAsyncHttpClient implements HttpClient { private static final ClientLogger LOGGER = new ClientLogger(NettyAsyncHttpClient.class); private static final byte[] EMPTY_BYTES = new byte[0]; + // Context key under which a caller may place an AtomicBoolean that this client sets to true when an interim + // "100 Continue" response is observed for the request. This is an opt-in contract usable by any caller that wants + // to learn whether the service engaged in the "Expect: 100-continue" handshake: the caller places a holder under + // this key before sending and reads it afterward. Callers must use this same key string. + private static final String EXPECT_CONTINUE_RECEIVED_KEY = "azure-http-client-expect-continue-received"; + final boolean disableBufferCopy; final boolean addProxyHandler; final reactor.netty.http.client.HttpClient nettyClient; @@ -136,14 +143,18 @@ public Mono send(HttpRequest request, Context context) { .map(timeoutDuration -> ((Duration) timeoutDuration).toMillis()) .orElse(null); ProgressReporter progressReporter = Contexts.with(context).getHttpRequestProgressReporter(); + AtomicBoolean expectContinueReceived = context.getData(EXPECT_CONTINUE_RECEIVED_KEY) + .filter(AtomicBoolean.class::isInstance) + .map(AtomicBoolean.class::cast) + .orElse(null); return attemptAsync(request, eagerlyReadResponse, ignoreResponseBody, headersEagerlyConverted, responseTimeout, - progressReporter, false); + progressReporter, expectContinueReceived, false); } private Mono attemptAsync(HttpRequest request, boolean eagerlyReadResponse, boolean ignoreResponseBody, boolean headersEagerlyConverted, Long responseTimeout, - ProgressReporter progressReporter, boolean proxyRetry) { + ProgressReporter progressReporter, AtomicBoolean expectContinueReceived, boolean proxyRetry) { Flux> nettyRequest = nettyClient.request(toReactorNettyHttpMethod(request.getHttpMethod())) .uri(request.getUrl().toString()) @@ -151,9 +162,11 @@ private Mono attemptAsync(HttpRequest request, boolean eagerlyRead .responseConnection(responseDelegate(request, disableBufferCopy, eagerlyReadResponse, ignoreResponseBody, headersEagerlyConverted)); - if (responseTimeout != null || progressReporter != null) { + // Attach the per-request context only when something needs it: a response-timeout override, a progress + // reporter, or a holder to record whether an interim "100 Continue" response was observed. + if (responseTimeout != null || progressReporter != null || expectContinueReceived != null) { nettyRequest = nettyRequest.contextWrite(ctx -> ctx.put(AzureNettyHttpClientContext.KEY, - new AzureNettyHttpClientContext(responseTimeout, progressReporter))); + new AzureNettyHttpClientContext(responseTimeout, progressReporter, expectContinueReceived))); } return nettyRequest.single().flatMap(responseAndHeaders -> { @@ -166,7 +179,7 @@ private Mono attemptAsync(HttpRequest request, boolean eagerlyRead } else { // Retry the request. return attemptAsync(request, eagerlyReadResponse, ignoreResponseBody, headersEagerlyConverted, - responseTimeout, progressReporter, true); + responseTimeout, progressReporter, expectContinueReceived, true); } } else { return Mono.just(response); @@ -174,7 +187,7 @@ private Mono attemptAsync(HttpRequest request, boolean eagerlyRead }) .onErrorResume(throwable -> shouldRetryProxyError(proxyRetry, throwable) ? attemptAsync(request, eagerlyReadResponse, ignoreResponseBody, headersEagerlyConverted, - responseTimeout, progressReporter, true) + responseTimeout, progressReporter, expectContinueReceived, true) : Mono.error(throwable)); } diff --git a/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/implementation/AzureNettyHttpClientContext.java b/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/implementation/AzureNettyHttpClientContext.java index 711aa3956672c..257444796faf5 100644 --- a/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/implementation/AzureNettyHttpClientContext.java +++ b/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/implementation/AzureNettyHttpClientContext.java @@ -5,6 +5,8 @@ import com.azure.core.util.ProgressReporter; +import java.util.concurrent.atomic.AtomicBoolean; + /** * A holder for all context that may be passed through the Netty HttpClient. */ @@ -13,15 +15,24 @@ public final class AzureNettyHttpClientContext { private final Long responseTimeoutOverride; private final ProgressReporter progressReporter; + // Optional caller-provided holder, set to true when an interim "100 Continue" response is observed for the + // request. The holder is placed in the request Context by a caller (for example a pipeline policy) that wants to + // observe the "Expect: 100-continue" handshake. May be null. + private final AtomicBoolean expectContinueReceivedHolder; + /** * Creates an instance of AzureNettyHttpClientContext. * * @param responseTimeoutOverride The response timeout override. * @param progressReporter The progress reporter. + * @param expectContinueReceivedHolder Optional caller-provided holder set to {@code true} when a + * {@code 100 Continue} response is observed. May be null. */ - public AzureNettyHttpClientContext(Long responseTimeoutOverride, ProgressReporter progressReporter) { + public AzureNettyHttpClientContext(Long responseTimeoutOverride, ProgressReporter progressReporter, + AtomicBoolean expectContinueReceivedHolder) { this.responseTimeoutOverride = responseTimeoutOverride; this.progressReporter = progressReporter; + this.expectContinueReceivedHolder = expectContinueReceivedHolder; } /** @@ -41,4 +52,14 @@ public Long getResponseTimeoutOverride() { public ProgressReporter getProgressReporter() { return progressReporter; } + + /** + * Gets the optional holder that is set to {@code true} when an interim {@code 100 Continue} response is observed + * for this request. May be null. + * + * @return The holder, or null if the caller did not supply one. + */ + public AtomicBoolean getExpectContinueReceivedHolder() { + return expectContinueReceivedHolder; + } } diff --git a/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/implementation/AzureSdkHandler.java b/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/implementation/AzureSdkHandler.java index f4600bc5fbc44..9286d34614a10 100644 --- a/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/implementation/AzureSdkHandler.java +++ b/sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/implementation/AzureSdkHandler.java @@ -12,11 +12,14 @@ import io.netty.channel.ChannelOutboundBuffer; import io.netty.channel.ChannelPromise; import io.netty.channel.FileRegion; +import io.netty.handler.codec.http.HttpResponse; +import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.LastHttpContent; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; /** * Class containing all ChannelHandler concepts that the Azure SDKs use. @@ -34,6 +37,7 @@ public final class AzureSdkHandler extends ChannelDuplexHandler { private final long writeTimeoutMillis; private final ProgressReporter progressReporter; + private final AtomicBoolean expectContinueReceivedHolder; private long lastWriteMillis; private long lastWriteProgress; private boolean writeTrackingStarted; @@ -65,6 +69,7 @@ public AzureSdkHandler(AzureNettyHttpClientContext context, long writeTimeoutMil long readTimeoutMillis) { this.writeTimeoutMillis = writeTimeoutMillis; this.progressReporter = (context != null) ? context.getProgressReporter() : null; + this.expectContinueReceivedHolder = (context != null) ? context.getExpectContinueReceivedHolder() : null; this.responseTimeoutMillis = (context != null && context.getResponseTimeoutOverride() != null) ? context.getResponseTimeoutOverride() : responseTimeoutMillis; @@ -215,6 +220,15 @@ private void disposeResponseTimeoutWatcher() { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { + // Observe the interim "100 Continue" response before it is consumed by Reactor Netty. This does not change + // how or when the request body is sent; it only records that the service engaged in the handshake so that + // calling code can see it. Scoped to the exact 100 status so final responses are never matched. + if (expectContinueReceivedHolder != null + && msg instanceof HttpResponse + && ((HttpResponse) msg).status().code() == HttpResponseStatus.CONTINUE.code()) { + expectContinueReceivedHolder.set(true); + } + if (responseTrackingStarted) { endResponseTracking(); startReadTracking(); diff --git a/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyExpectContinueObservationTests.java b/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyExpectContinueObservationTests.java new file mode 100644 index 0000000000000..9d0f375ccb75a --- /dev/null +++ b/sdk/core/azure-core-http-netty/src/test/java/com/azure/core/http/netty/NettyExpectContinueObservationTests.java @@ -0,0 +1,188 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.core.http.netty; + +import com.azure.core.http.HttpClient; +import com.azure.core.http.HttpHeaderName; +import com.azure.core.http.HttpMethod; +import com.azure.core.http.HttpRequest; +import com.azure.core.http.HttpResponse; +import com.azure.core.util.Context; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.URI; +import java.net.URL; +import java.nio.charset.StandardCharsets; +import java.util.Locale; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies that the Netty client surfaces the observation of an interim {@code 100 Continue} response through an + * {@link AtomicBoolean} holder the caller places in the request {@link Context}. The holder is keyed by an opt-in + * string contract any caller may use; the Netty client sets it to {@code true} when the service sends the interim + * response. + */ +public class NettyExpectContinueObservationTests { + private static final byte[] BODY = new byte[1024]; + + @ParameterizedTest + @ValueSource(booleans = { true, false }) + public void interimContinueIsSurfacedThroughContextHolder(boolean sync) throws Exception { + try (ContinueTestServer server = new ContinueTestServer(true)) { + server.start(); + AtomicBoolean holder = sendWithHolder(server, sync); + assertTrue(server.awaitRequest(), "request never reached the server"); + assertTrue(holder.get(), "100 Continue was not surfaced through the context holder"); + } + } + + @ParameterizedTest + @ValueSource(booleans = { true, false }) + public void holderStaysFalseWhenServerSendsNoContinue(boolean sync) throws Exception { + try (ContinueTestServer server = new ContinueTestServer(false)) { + server.start(); + AtomicBoolean holder = sendWithHolder(server, sync); + assertTrue(server.awaitRequest(), "request never reached the server"); + assertFalse(holder.get(), "holder should stay false when no 100 Continue is sent"); + } + } + + // Must match the private key NettyAsyncHttpClient reads for the opt-in 100-continue observation contract. + private static final String EXPECT_CONTINUE_RECEIVED_KEY = "azure-http-client-expect-continue-received"; + + private static AtomicBoolean sendWithHolder(ContinueTestServer server, boolean sync) throws IOException { + HttpClient client = new NettyAsyncHttpClientBuilder().build(); + AtomicBoolean holder = new AtomicBoolean(false); + Context context = new Context(EXPECT_CONTINUE_RECEIVED_KEY, holder); + + HttpRequest request = new HttpRequest(HttpMethod.PUT, server.url()).setBody(BODY); + request.getHeaders().set(HttpHeaderName.EXPECT, "100-continue"); + + HttpResponse response = sync ? client.sendSync(request, context) : client.send(request, context).block(); + assertEquals(201, response.getStatusCode()); + response.close(); + return holder; + } + + /** + * Minimal HTTP/1.1 server that optionally answers the 100-continue handshake, then reads the body and returns 201. + */ + private static final class ContinueTestServer implements AutoCloseable { + private final ServerSocket serverSocket; + private final boolean sendContinue; + private final CountDownLatch requestHandled = new CountDownLatch(1); + private volatile Thread thread; + + ContinueTestServer(boolean sendContinue) throws IOException { + this.serverSocket = new ServerSocket(0); + this.sendContinue = sendContinue; + } + + URL url() throws IOException { + return URI.create("http://localhost:" + serverSocket.getLocalPort() + "/resource").toURL(); + } + + void start() { + thread = new Thread(this::serve, "continue-test-server"); + thread.setDaemon(true); + thread.start(); + } + + boolean awaitRequest() throws InterruptedException { + return requestHandled.await(30, TimeUnit.SECONDS); + } + + private void serve() { + try (Socket socket = serverSocket.accept()) { + socket.setTcpNoDelay(true); + InputStream in = socket.getInputStream(); + OutputStream out = socket.getOutputStream(); + + Parsed parsed = readHeaders(in); + + if (sendContinue && parsed.expectHeader != null) { + out.write("HTTP/1.1 100 Continue\r\n\r\n".getBytes(StandardCharsets.UTF_8)); + out.flush(); + } + + readBody(in, parsed.contentLength); + + out.write("HTTP/1.1 201 Created\r\nContent-Length: 0\r\n\r\n".getBytes(StandardCharsets.UTF_8)); + out.flush(); + } catch (IOException ex) { + // The test assertions report the failure. + } finally { + requestHandled.countDown(); + } + } + + private static Parsed readHeaders(InputStream in) throws IOException { + ByteArrayOutputStream headerBytes = new ByteArrayOutputStream(); + int consecutiveNewlines = 0; + int b; + while (consecutiveNewlines < 2 && (b = in.read()) != -1) { + headerBytes.write(b); + if (b == '\n') { + consecutiveNewlines++; + } else if (b != '\r') { + consecutiveNewlines = 0; + } + } + + Parsed parsed = new Parsed(); + for (String line : new String(headerBytes.toByteArray(), StandardCharsets.UTF_8).split("\r\n")) { + int colon = line.indexOf(':'); + if (colon < 0) { + continue; + } + String name = line.substring(0, colon).trim().toLowerCase(Locale.ROOT); + String value = line.substring(colon + 1).trim(); + if ("expect".equals(name)) { + parsed.expectHeader = value.toLowerCase(Locale.ROOT); + } else if ("content-length".equals(name)) { + parsed.contentLength = Integer.parseInt(value); + } + } + return parsed; + } + + private static void readBody(InputStream in, int remaining) throws IOException { + byte[] buffer = new byte[Math.max(remaining, 1)]; + int read = 0; + while (read < remaining) { + int count = in.read(buffer, 0, Math.min(buffer.length, remaining - read)); + if (count < 0) { + break; + } + read += count; + } + } + + @Override + public void close() throws IOException { + serverSocket.close(); + if (thread != null) { + thread.interrupt(); + } + } + + private static final class Parsed { + private String expectHeader; + private int contentLength; + } + } +}