[Core] Honour Expect: 100-continue in the Netty HTTP Client

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
 

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,24 +143,30 @@ public Mono<HttpResponse> 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<HttpResponse> attemptAsync(HttpRequest request, boolean eagerlyReadResponse,
         boolean ignoreResponseBody, boolean headersEagerlyConverted, Long responseTimeout,
-        ProgressReporter progressReporter, boolean proxyRetry) {
+        ProgressReporter progressReporter, AtomicBoolean expectContinueReceived, boolean proxyRetry) {
         Flux<Tuple2<HttpResponse, HttpHeaders>> nettyRequest
             = nettyClient.request(toReactorNettyHttpMethod(request.getHttpMethod()))
                 .uri(request.getUrl().toString())
                 .send(bodySendDelegate(request))
                 .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,15 +179,15 @@ private Mono<HttpResponse> 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);
             }
         })
             .onErrorResume(throwable -> shouldRetryProxyError(proxyRetry, throwable)
                 ? attemptAsync(request, eagerlyReadResponse, ignoreResponseBody, headersEagerlyConverted,
-                    responseTimeout, progressReporter, true)
+                    responseTimeout, progressReporter, expectContinueReceived, true)
                 : Mono.error(throwable));
     }
 

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;
+    }
 }

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();

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;
+        }
+    }
+}