Remove Void Responses from being Eagerly Read

sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/JdkHttpClient.java

@@ -8,6 +8,7 @@
 import com.azure.core.http.HttpHeaders;
 import com.azure.core.http.HttpRequest;
 import com.azure.core.http.HttpResponse;
+import com.azure.core.http.jdk.httpclient.implementation.BodyIgnoringSubscriber;
 import com.azure.core.util.Context;
 import com.azure.core.util.Contexts;
 import com.azure.core.util.CoreUtils;
@@ -20,7 +21,6 @@
 import reactor.core.publisher.Mono;
 
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.UncheckedIOException;
 import java.net.URISyntaxException;
 import java.util.List;
@@ -37,6 +37,9 @@
  */
 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 static final byte[] IGNORED_BODY = new byte[0];
 
     private final java.net.http.HttpClient jdkHttpClient;
 
@@ -61,11 +64,24 @@ public Mono<HttpResponse> send(HttpRequest request) {
 
     @Override
     public Mono<HttpResponse> send(HttpRequest request, Context context) {
-        boolean eagerlyReadResponse = (boolean) context.getData("azure-eagerly-read-response").orElse(false);
+        boolean eagerlyReadResponse = (boolean) context.getData(AZURE_EAGERLY_READ_RESPONSE).orElse(false);
+        boolean ignoreResponseBody = (boolean) context.getData(AZURE_IGNORE_RESPONSE_BODY).orElse(false);
 
         return Mono.fromCallable(() -> toJdkHttpRequest(request, context))
             .flatMap(jdkRequest -> Mono.fromCompletionStage(jdkHttpClient.sendAsync(jdkRequest, ofPublisher()))
                 .flatMap(jdKResponse -> {
+                    // Ignoring the response body takes precedent over eagerly reading the response body.
+                    // Both should never be true at the same time but this is acts as a safeguard.
+                    if (ignoreResponseBody) {
+                        HttpHeaders headers = fromJdkHttpHeaders(jdKResponse.headers());
+                        int statusCode = jdKResponse.statusCode();
+
+                        return JdkFlowAdapter.flowPublisherToFlux(jdKResponse.body())
+                            .ignoreElements()
+                            .then(Mono.fromSupplier(() ->
+                                new JdkHttpResponseSync(request, statusCode, headers, IGNORED_BODY)));
+                    }
+
                     if (eagerlyReadResponse) {
                         HttpHeaders headers = fromJdkHttpHeaders(jdKResponse.headers());
                         int statusCode = jdKResponse.statusCode();
@@ -82,16 +98,24 @@ public Mono<HttpResponse> 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 eagerlyReadResponse = (boolean) context.getData(AZURE_EAGERLY_READ_RESPONSE).orElse(false);
+        boolean ignoreResponseBody = (boolean) context.getData(AZURE_IGNORE_RESPONSE_BODY).orElse(false);
 
         java.net.http.HttpRequest jdkRequest = toJdkHttpRequest(request, context);
         try {
-            if (eagerlyReadResponse) {
+            // Ignoring the response body takes precedent over eagerly reading the response body.
+            // Both should never be true at the same time but this is acts as a safeguard.
+            if (ignoreResponseBody) {
+                java.net.http.HttpResponse<Void> jdKResponse = jdkHttpClient.send(jdkRequest,
+                    responseInfo -> new BodyIgnoringSubscriber(LOGGER));
+                return new JdkHttpResponseSync(request, jdKResponse.statusCode(),
+                    fromJdkHttpHeaders(jdKResponse.headers()), IGNORED_BODY);
+            } else if (eagerlyReadResponse) {
                 java.net.http.HttpResponse<byte[]> jdKResponse = jdkHttpClient.send(jdkRequest, ofByteArray());
-                return new JdkHttpResponseSync(request, jdKResponse.statusCode(), fromJdkHttpHeaders(jdKResponse.headers()), jdKResponse.body());
+                return new JdkHttpResponseSync(request, jdKResponse.statusCode(),
+                    fromJdkHttpHeaders(jdKResponse.headers()), jdKResponse.body());
             } else {
-                java.net.http.HttpResponse<InputStream> jdKResponse = jdkHttpClient.send(jdkRequest, ofInputStream());
-                return new JdkHttpResponseSync(request, jdKResponse);
+                return new JdkHttpResponseSync(request, jdkHttpClient.send(jdkRequest, ofInputStream()));
             }
         } catch (IOException e) {
             throw LOGGER.logExceptionAsError(new UncheckedIOException(e));

sdk/core/azure-core-http-jdk-httpclient/src/main/java/com/azure/core/http/jdk/httpclient/implementation/BodyIgnoringSubscriber.java

@@ -0,0 +1,68 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.http.jdk.httpclient.implementation;
+
+import com.azure.core.http.HttpClient;
+import com.azure.core.util.logging.ClientLogger;
+import com.azure.core.util.logging.LogLevel;
+
+import java.net.http.HttpResponse;
+import java.nio.ByteBuffer;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Flow;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Implementation of a {@link HttpResponse.BodySubscriber} that ignores the response body.
+ * <p>
+ * This is used when the {@link HttpClient} is told to ignore the response body, used when the returned body value is
+ * {@code void} or {@code Void}.
+ * <p>
+ * This will log a warning message if a response body is received to indicate that there was a bug either in determining
+ * that the response body should be ignored, the Swagger indicated no response body would be received but was, or that
+ * the server sent a response body when it shouldn't.
+ */
+public final class BodyIgnoringSubscriber implements HttpResponse.BodySubscriber<Void> {
+    private final CompletableFuture<Void> completableFuture;
+    private final ClientLogger logger;
+    private final AtomicBoolean subscribed = new AtomicBoolean();
+
+    public BodyIgnoringSubscriber(ClientLogger logger) {
+        this.completableFuture = new CompletableFuture<>();
+        this.logger = logger;
+    }
+
+    @Override
+    public CompletionStage<Void> getBody() {
+        return completableFuture;
+    }
+
+    @Override
+    public void onSubscribe(Flow.Subscription subscription) {
+        if (!subscribed.compareAndSet(false, true)) {
+            // Only can have one subscription.
+            subscription.cancel();
+        } else {
+            subscription.request(Long.MAX_VALUE);
+        }
+    }
+
+    @Override
+    public void onNext(List<ByteBuffer> item) {
+        logger.log(LogLevel.WARNING, () -> "Received HTTP response body when one wasn't expected. "
+            + "Response body will be ignored as directed.");
+    }
+
+    @Override
+    public void onError(Throwable throwable) {
+        completableFuture.completeExceptionally(throwable);
+    }
+
+    @Override
+    public void onComplete() {
+        completableFuture.complete(null);
+    }
+}

sdk/core/azure-core-http-netty/src/main/java/com/azure/core/http/netty/NettyAsyncHttpClient.java

@@ -26,6 +26,7 @@
 import com.azure.core.util.Contexts;
 import com.azure.core.util.ProgressReporter;
 import com.azure.core.util.logging.ClientLogger;
+import com.azure.core.util.logging.LogLevel;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.EventLoopGroup;
@@ -51,16 +52,17 @@
 import java.nio.file.StandardOpenOption;
 import java.time.Duration;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
 
 import static com.azure.core.http.netty.implementation.Utility.closeConnection;
 
 /**
  * This class provides a Netty-based implementation for the {@link HttpClient} interface. Creating an instance of this
  * class can be achieved by using the {@link NettyAsyncHttpClientBuilder} class, which offers Netty-specific API for
- * features such as {@link NettyAsyncHttpClientBuilder#eventLoopGroup(EventLoopGroup) thread pooling}, {@link
- * NettyAsyncHttpClientBuilder#wiretap(boolean) wiretapping}, {@link NettyAsyncHttpClientBuilder#proxy(ProxyOptions)
- * setProxy configuration}, and much more.
+ * features such as {@link NettyAsyncHttpClientBuilder#eventLoopGroup(EventLoopGroup) thread pooling},
+ * {@link NettyAsyncHttpClientBuilder#wiretap(boolean) wiretapping},
+ * {@link NettyAsyncHttpClientBuilder#proxy(ProxyOptions) setProxy configuration}, and much more.
  *
  * @see HttpClient
  * @see NettyAsyncHttpClientBuilder
@@ -70,6 +72,7 @@ class NettyAsyncHttpClient implements HttpClient {
     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";
 
@@ -109,24 +112,24 @@ public Mono<HttpResponse> 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 effectiveEagerlyReadResponse = (boolean) context.getData(AZURE_EAGERLY_READ_RESPONSE).orElse(false);
-        long effectiveResponseTimeout = context.getData(AZURE_RESPONSE_TIMEOUT)
+        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)
             .filter(timeoutDuration -> timeoutDuration instanceof Duration)
             .map(timeoutDuration -> ((Duration) timeoutDuration).toMillis())
             .orElse(this.responseTimeout);
-        boolean effectiveHeadersEagerlyConverted = (boolean) context.getData(AZURE_EAGERLY_CONVERT_HEADERS)
-            .orElse(false);
 
         return nettyClient
             .doOnRequest((r, connection) -> addRequestHandlers(connection, context))
-            .doAfterRequest((r, connection) -> doAfterRequest(connection, effectiveResponseTimeout))
+            .doAfterRequest((r, connection) -> doAfterRequest(connection, responseTimeout))
             .doOnResponse((response, connection) -> addReadTimeoutHandler(connection, readTimeout))
             .doAfterResponseSuccess((response, connection) -> removeReadTimeoutHandler(connection))
             .request(HttpMethod.valueOf(request.getHttpMethod().toString()))
             .uri(request.getUrl().toString())
             .send(bodySendDelegate(request))
-            .responseConnection(responseDelegate(request, disableBufferCopy, effectiveEagerlyReadResponse,
-                effectiveHeadersEagerlyConverted))
+            .responseConnection(responseDelegate(request, disableBufferCopy, eagerlyReadResponse, ignoreResponseBody,
+                headersEagerlyConverted))
             .single()
             .onErrorMap(throwable -> {
                 // The exception was an SSLException that was caused by a failure to connect to a proxy.
@@ -247,14 +250,32 @@ private static NettyOutbound sendInputStream(NettyOutbound reactorNettyOutbound,
      * @param restRequest the Rest request whose response this delegate handles
      * @param disableBufferCopy Flag indicating if the network response shouldn't be buffered.
      * @param eagerlyReadResponse Flag indicating if the network response should be eagerly read into memory.
+     * @param ignoreResponseBody Flag indicating if the network response should be ignored.
      * @param headersEagerlyConverted Flag indicating if the Netty HttpHeaders should be eagerly converted to Azure Core
      * HttpHeaders.
      * @return a delegate upon invocation setup Rest response object
      */
     private static BiFunction<HttpClientResponse, Connection, Publisher<HttpResponse>> responseDelegate(
-        HttpRequest restRequest, boolean disableBufferCopy, boolean eagerlyReadResponse,
+        HttpRequest restRequest, boolean disableBufferCopy, boolean eagerlyReadResponse, boolean ignoreResponseBody,
         boolean headersEagerlyConverted) {
         return (reactorNettyResponse, reactorNettyConnection) -> {
+            // Ignoring the response body takes precedent over eagerly reading the response body.
+            // Both should never be true at the same time but this is acts as a safeguard.
+            if (ignoreResponseBody) {
+                AtomicBoolean firstNext = new AtomicBoolean(true);
+                return reactorNettyConnection.inbound().receive()
+                    .doOnNext(ignored -> {
+                        if (!firstNext.compareAndSet(true, false)) {
+                            LOGGER.log(LogLevel.WARNING, () -> "Received HTTP response body when one wasn't expected. "
+                                + "Response body will be ignored as directed.");
+                        }
+                    })
+                    .ignoreElements()
+                    .doFinally(ignored -> closeConnection(reactorNettyConnection))
+                    .then(Mono.fromSupplier(() -> new NettyAsyncHttpBufferedResponse(reactorNettyResponse, restRequest,
+                        EMPTY_BYTES, headersEagerlyConverted)));
+            }
+
             /*
              * If the response is being eagerly read into memory the flag for buffer copying can be ignored as the
              * response MUST be deeply copied to ensure it can safely be used downstream.