[Core] Honour Expect: 100-continue in the Vert.x HTTP client (#50312)

* Honour Expect: 100-continue in the Vert.x HTTP client

sendBody always wrote the headers and the body together, so a request
carrying Expect: 100-continue behaved exactly like one without it and the
service had no opportunity to reject it before receiving the body.

Vert.x already exposes the two pieces needed. When the header is present,
send the head and write the body from continueHandler; otherwise keep the
existing behaviour. The body writing is now shared by both paths, so this
covers in-memory and streaming bodies alike.

Adds wire level tests that drive the client against a socket which delays
its 100 Continue, asserting that no body bytes arrive before the response
and that the full body arrives after it.

* Avoid the deprecated URL constructor in the test

The core modules compile with -Werror, and URL(String) is deprecated as
of Java 20, so the test failed to compile on the JDK 25 build agents.

* Address review feedback

Send the body after a one second fallback if the service never answers
the expectation, so a server that ignores it cannot stall the request
until the response timeout. The body is guarded so it is written once
whether the interim response or the fallback comes first, and the timer
is cancelled on continue.

Expect is a comma separated list and values may carry whitespace, so
compare each expectation rather than the header as a whole.

The test server now reads for the whole settle window rather than taking
an available() snapshot after a sleep. Adds cases for the header spelling
variants and for a service that ignores the expectation entirely.

* Do not write the body once the exchange is terminal

The fallback was cancelled only from continueHandler, so a service that
rejected the headers with a final status left the timer armed. It would
then write onto a finished exchange, and for a streaming body would
subscribe the source needlessly.

Cancel the fallback and claim the write when the response future reaches
any terminal state, so neither the timer nor a late interim response can
send the body afterwards. Adds a regression test for a service that
answers the headers with 401 and never reads the body.

Author: gunjansingh-msft

sdk/core/azure-core-http-vertx/CHANGELOG.md

@@ -3,6 +3,11 @@
 ## 1.2.0-beta.1 (Unreleased)
 
 ### Features Added
+- Requests carrying the `Expect: 100-continue` header now perform the handshake: the headers are sent with
+  `sendHead()` and the request body is written from `continueHandler()`, rather than sending headers and body
+  together.
+  If the service does not answer the expectation the body is sent after a one second fallback, so the request does
+  not stall.
 
 ### Breaking Changes
 

sdk/core/azure-core-http-vertx/src/main/java/com/azure/core/http/vertx/VertxHttpClient.java

@@ -26,6 +26,7 @@
 import com.azure.core.util.logging.ClientLogger;
 import io.vertx.core.Future;
 import io.vertx.core.Promise;
+import io.vertx.core.Vertx;
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.http.HttpClientOptions;
 import io.vertx.core.http.HttpClientRequest;
@@ -39,6 +40,7 @@
 import java.io.UncheckedIOException;
 import java.time.Duration;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
@@ -49,6 +51,10 @@
 class VertxHttpClient implements HttpClient {
     private static final ClientLogger LOGGER = new ClientLogger(VertxHttpClient.class);
 
+    // How long to wait for an interim 100 Continue before sending the body anyway, matching the defaults used by
+    // other HTTP stacks. RFC 9110 requires a client not to wait indefinitely.
+    private static final Duration EXPECT_CONTINUE_TIMEOUT = Duration.ofSeconds(1);
+
     final io.vertx.core.http.HttpClient client;
     final HttpClientOptions buildOptions;
     private final Duration responseTimeout;
@@ -183,26 +189,89 @@ private void sendBody(ContextView contextView, HttpRequest azureRequest, Progres
 
         if (body == null) {
             vertxRequest.send().onFailure(promise::fail);
+            return;
+        }
+
+        if (expectsContinue(azureRequest.getHeaders().getValue(HttpHeaderName.EXPECT))) {
+            // Send the headers and hold the body until the service accepts the expectation. If it answers with a
+            // final status instead, that response completes the request and the body is never written.
+            AtomicBoolean bodyWritten = new AtomicBoolean();
+            Runnable writeOnce = () -> {
+                if (bodyWritten.compareAndSet(false, true)) {
+                    writeBody(contextView, azureRequest, progressReporter, vertxRequest, promise, body, true);
+                }
+            };
+
+            // A service may ignore the expectation entirely and wait for the body, so the request cannot depend on
+            // the interim response arriving. RFC 9110 allows the content to be sent after a short wait.
+            io.vertx.core.Context vertxContext = Vertx.currentContext();
+            Long fallbackTimerId = vertxContext == null
+                ? null
+                : vertxContext.owner().setTimer(EXPECT_CONTINUE_TIMEOUT.toMillis(), ignored -> writeOnce.run());
+            Runnable cancelFallback = () -> {
+                if (fallbackTimerId != null && vertxContext != null) {
+                    vertxContext.owner().cancelTimer(fallbackTimerId);
+                }
+            };
+
+            // Once the exchange is terminal, for example the service rejected the headers outright or sending them
+            // failed, there is nothing left to send. Claim the write so neither the fallback nor a late interim
+            // response can subscribe the body onto a finished request.
+            promise.future().onComplete(ignored -> {
+                cancelFallback.run();
+                bodyWritten.set(true);
+            });
+
+            vertxRequest.continueHandler(ignored -> {
+                cancelFallback.run();
+                writeOnce.run();
+            });
+            vertxRequest.sendHead().onFailure(promise::fail);
         } else {
-            BinaryDataContent bodyContent = BinaryDataHelper.getContent(body);
-            if (bodyContent instanceof ByteArrayContent
-                || bodyContent instanceof ByteBufferContent
-                || bodyContent instanceof StringContent
-                || bodyContent instanceof SerializableContent) {
-                // This cannot produce a NullPointerException for the BinaryDataContent types that trigger this as they
-                // are in-memory and have a length.
-                long contentLength = bodyContent.getLength();
-                vertxRequest.send(Buffer.buffer(bodyContent.toBytes()))
-                    .onSuccess(ignored -> reportProgress(contentLength, progressReporter))
-                    .onFailure(promise::fail);
-            } else {
-                // Right now both Flux<ByteBuffer> and InputStream bodies are being handled reactively.
-                azureRequest.getBody()
-                    .subscribe(new VertxRequestWriteSubscriber(vertxRequest::exceptionHandler,
-                        vertxRequest::drainHandler, vertxRequest::write, vertxRequest::writeQueueFull,
-                        vertxRequest::reset, vertxRequest::end, promise, progressReporter, contextView));
+            writeBody(contextView, azureRequest, progressReporter, vertxRequest, promise, body, false);
+        }
+    }
+
+    /*
+     * Expect is a comma separated list of expectations and values may carry surrounding whitespace, so the header
+     * cannot be compared as a whole.
+     */
+    private static boolean expectsContinue(String expectHeader) {
+        if (expectHeader == null) {
+            return false;
+        }
+
+        for (String expectation : expectHeader.split(",")) {
+            if ("100-continue".equalsIgnoreCase(expectation.trim())) {
+                return true;
             }
         }
+
+        return false;
+    }
+
+    private void writeBody(ContextView contextView, HttpRequest azureRequest, ProgressReporter progressReporter,
+        HttpClientRequest vertxRequest, Promise<HttpResponse> promise, BinaryData body, boolean headAlreadySent) {
+        BinaryDataContent bodyContent = BinaryDataHelper.getContent(body);
+        if (bodyContent instanceof ByteArrayContent
+            || bodyContent instanceof ByteBufferContent
+            || bodyContent instanceof StringContent
+            || bodyContent instanceof SerializableContent) {
+            // This cannot produce a NullPointerException for the BinaryDataContent types that trigger this as they
+            // are in-memory and have a length.
+            long contentLength = bodyContent.getLength();
+            Buffer buffer = Buffer.buffer(bodyContent.toBytes());
+            // Once the head has been sent the body must be written onto the open request rather than sent as a
+            // whole new request.
+            Future<Void> written = headAlreadySent ? vertxRequest.end(buffer) : vertxRequest.send(buffer).mapEmpty();
+            written.onSuccess(ignored -> reportProgress(contentLength, progressReporter)).onFailure(promise::fail);
+        } else {
+            // Right now both Flux<ByteBuffer> and InputStream bodies are being handled reactively.
+            azureRequest.getBody()
+                .subscribe(new VertxRequestWriteSubscriber(vertxRequest::exceptionHandler, vertxRequest::drainHandler,
+                    vertxRequest::write, vertxRequest::writeQueueFull, vertxRequest::reset, vertxRequest::end, promise,
+                    progressReporter, contextView));
+        }
     }
 
     private static void reportProgress(long progress, ProgressReporter progressReporter) {