HttpPipeline sync refactor (Azure#29049)

Author: samvaity

eng/code-quality-reports/src/main/resources/spotbugs/spotbugs-exclude.xml

@@ -385,7 +385,9 @@
   <Match>
     <Or>
       <Class name="com.azure.core.http.HttpPipelineNextPolicy"/>
+      <Class name="com.azure.core.http.HttpPipelineNextSyncPolicy"/>
       <Class name="com.azure.core.http.HttpRequest"/>
+      <Class name="com.azure.core.implementation.http.HttpPipelineCallState"/>
     </Or>
     <Bug pattern="CN_IMPLEMENTS_CLONE_BUT_NOT_CLONEABLE"/>
   </Match>

sdk/core/azure-core/CHANGELOG.md

@@ -4,6 +4,12 @@
 
 ### Features Added
 
+- Added support for sending synchronous requests using `sendSync` in `HttpPipeline`:
+  - Added `HttpPipelinePolicy.processSync(HttpPipelineCallContext context, HttpPipelineNextSyncPolicy next)` to allow processing policies synchronously.
+  - Added `HttpPipelineSyncPolicy` to represent synchronous `HttpPipelinePolicy`.
+  - Added `HttpPipelineNextSyncPolicy` to invoke the next synchronous policy in pipeline. to process synchronous policy pipeline.
+  - Added `HttpPipelineCallState` to maintain request specific pipeline and contextual data.
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/core/azure-core/src/main/java/com/azure/core/http/HttpClient.java

@@ -31,6 +31,17 @@ default Mono<HttpResponse> send(HttpRequest request, Context context) {
         return send(request);
     }
 
+    /**
+     * Sends the provided request synchronously with contextual information.
+     *
+     * @param request The HTTP request to send.
+     * @param context Contextual information about the request.
+     * @return The response.
+     */
+    default HttpResponse sendSync(HttpRequest request, Context context) {
+        return send(request, context).block();
+    }
+
     /**
      * Creates a new {@link HttpClient} instance.
      *

sdk/core/azure-core/src/main/java/com/azure/core/http/HttpPipeline.java

@@ -4,6 +4,7 @@
 package com.azure.core.http;
 
 import com.azure.core.http.policy.HttpPipelinePolicy;
+import com.azure.core.implementation.http.HttpPipelineCallState;
 import com.azure.core.util.Context;
 import reactor.core.publisher.Mono;
 
@@ -89,6 +90,7 @@ public Mono<HttpResponse> send(HttpRequest request, Context data) {
         return this.send(new HttpPipelineCallContext(request, data));
     }
 
+
     /**
      * Sends the context (containing an HTTP request) through pipeline.
      *
@@ -99,8 +101,23 @@ public Mono<HttpResponse> send(HttpRequest request, Context data) {
     public Mono<HttpResponse> send(HttpPipelineCallContext context) {
         // Return deferred to mono for complete lazy behaviour.
         return Mono.defer(() -> {
-            HttpPipelineNextPolicy next = new HttpPipelineNextPolicy(this, context);
+            HttpPipelineNextPolicy next =
+                new HttpPipelineNextPolicy(new HttpPipelineCallState(this, context));
             return next.process();
         });
     }
+
+    /**
+     * Wraps the request in a context with additional metadata and sends it through the pipeline.
+     *
+     * @param request THe HTTP request to send.
+     * @param data Additional metadata to pass along with the request.
+     * @return A publisher upon subscription flows the context through policies, sends the request, and emits response
+     * upon completion.
+     */
+    public HttpResponse sendSync(HttpRequest request, Context data) {
+        HttpPipelineNextSyncPolicy next = new HttpPipelineNextSyncPolicy(
+            new HttpPipelineCallState(this, new HttpPipelineCallContext(request, data)));
+        return next.processSync();
+    }
 }

sdk/core/azure-core/src/main/java/com/azure/core/http/HttpPipelineNextPolicy.java

@@ -4,28 +4,41 @@
 package com.azure.core.http;
 
 import com.azure.core.http.policy.HttpPipelinePolicy;
+import com.azure.core.implementation.http.HttpPipelineCallState;
+import com.azure.core.util.logging.ClientLogger;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 /**
  * A type that invokes next policy in the pipeline.
  */
 public class HttpPipelineNextPolicy {
-    private final HttpPipeline pipeline;
-    private final HttpPipelineCallContext context;
-    private int currentPolicyIndex;
+    private static final ClientLogger LOGGER = new ClientLogger(HttpPipelineNextPolicy.class);
+    private final HttpPipelineCallState state;
+    private final boolean originatedFromSyncPolicy;
 
     /**
      * Package Private ctr.
      *
      * Creates HttpPipelineNextPolicy.
      *
-     * @param pipeline the pipeline
-     * @param context the request-response context
+     * @param state the pipeline call state.
      */
-    HttpPipelineNextPolicy(final HttpPipeline pipeline, HttpPipelineCallContext context) {
-        this.pipeline = pipeline;
-        this.context = context;
-        this.currentPolicyIndex = -1;
+    HttpPipelineNextPolicy(HttpPipelineCallState state) {
+        this.state = state;
+        this.originatedFromSyncPolicy = false;
+    }
+
+    /**
+     * Package Private ctr.
+     * Creates HttpPipelineNextPolicy.
+     *
+     * @param state the pipeline call state.
+     * @param originatedFromSyncPolicy boolean to indicate if the next policy originated from sync call stack.
+     */
+    HttpPipelineNextPolicy(HttpPipelineCallState state, boolean originatedFromSyncPolicy) {
+        this.state = state;
+        this.originatedFromSyncPolicy = originatedFromSyncPolicy;
     }
 
     /**
@@ -34,16 +47,25 @@ public class HttpPipelineNextPolicy {
      * @return A publisher which upon subscription invokes next policy and emits response from the policy.
      */
     public Mono<HttpResponse> process() {
-        final int size = this.pipeline.getPolicyCount();
-        if (this.currentPolicyIndex > size) {
-            return Mono.error(new IllegalStateException("There is no more policies to execute."));
-        }
-
-        this.currentPolicyIndex++;
-        if (this.currentPolicyIndex == size) {
-            return this.pipeline.getHttpClient().send(this.context.getHttpRequest(), this.context.getContext());
+        if (originatedFromSyncPolicy && !Schedulers.isInNonBlockingThread()) {
+            // Pipeline executes in synchronous style. We most likely got here via default implementation in the
+            // HttpPipelinePolicy.processSynchronously so go back to sync style here.
+            // Don't do this on non-blocking threads.
+            return Mono.fromCallable(() -> new HttpPipelineNextSyncPolicy(state).processSync());
         } else {
-            return this.pipeline.getPolicy(this.currentPolicyIndex).process(this.context, this);
+            if (originatedFromSyncPolicy) {
+                LOGGER.warning("The pipeline switched from synchronous to asynchronous."
+                    + "Check if {} does not override HttpPipelinePolicy.processSync",
+                    this.state.getCurrentPolicy().getClass().getSimpleName());
+            }
+
+            HttpPipelinePolicy nextPolicy = state.getNextPolicy();
+            if (nextPolicy == null) {
+                return this.state.getPipeline().getHttpClient().send(
+                    this.state.getCallContext().getHttpRequest(), this.state.getCallContext().getContext());
+            } else {
+                return nextPolicy.process(this.state.getCallContext(), this);
+            }
         }
     }
 
@@ -54,8 +76,6 @@ public Mono<HttpResponse> process() {
      */
     @Override
     public HttpPipelineNextPolicy clone() {
-        HttpPipelineNextPolicy cloned = new HttpPipelineNextPolicy(this.pipeline, this.context);
-        cloned.currentPolicyIndex = this.currentPolicyIndex;
-        return cloned;
+        return new HttpPipelineNextPolicy(this.state.clone(), this.originatedFromSyncPolicy);
     }
 }

sdk/core/azure-core/src/main/java/com/azure/core/http/HttpPipelineNextSyncPolicy.java

@@ -0,0 +1,65 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.http;
+
+import com.azure.core.http.policy.HttpPipelinePolicy;
+import com.azure.core.implementation.http.HttpPipelineCallState;
+import com.azure.core.implementation.http.HttpPipelineNextSyncPolicyHelper;
+
+/**
+ * A type that invokes next policy in the pipeline.
+ */
+public class HttpPipelineNextSyncPolicy {
+    private final HttpPipelineCallState state;
+
+    static {
+        HttpPipelineNextSyncPolicyHelper.setAccessor(HttpPipelineNextSyncPolicy::toAsyncPolicy);
+    }
+
+    /**
+     * Package Private ctr.
+     * Creates HttpPipelineNextPolicy.
+     *
+     * @param state the pipeline call state.
+     */
+    HttpPipelineNextSyncPolicy(HttpPipelineCallState state) {
+        this.state = state;
+    }
+
+    /**
+     * Invokes the next {@link HttpPipelinePolicy}.
+     *
+     * @return The response.
+     */
+    public HttpResponse processSync() {
+        HttpPipelinePolicy nextPolicy = state.getNextPolicy();
+        if (nextPolicy == null) {
+            return this.state.getPipeline().getHttpClient().sendSync(
+                this.state.getCallContext().getHttpRequest(), this.state.getCallContext().getContext());
+        } else {
+            return nextPolicy.processSync(this.state.getCallContext(), this);
+        }
+    }
+
+    /**
+     * Creates a new instance of this instance.
+     *
+     * @return A new instance of this next pipeline sync policy.
+     */
+    @Override
+    public HttpPipelineNextSyncPolicy clone() {
+        return new HttpPipelineNextSyncPolicy(this.state.clone());
+    }
+
+    /**
+     * Method to convert a {@link HttpPipelineNextSyncPolicy} to a {@link HttpPipelineNextPolicy} for supporting the
+     * default implementation of
+     * {@link HttpPipelinePolicy#processSync(HttpPipelineCallContext, HttpPipelineNextSyncPolicy)}.
+     *
+     * @return the converted {@link HttpPipelineNextSyncPolicy}.
+     */
+    HttpPipelineNextPolicy toAsyncPolicy() {
+        return new HttpPipelineNextPolicy(this.state, true);
+    }
+}

sdk/core/azure-core/src/main/java/com/azure/core/http/policy/HttpPipelinePolicy.java

@@ -6,8 +6,10 @@
 import com.azure.core.http.HttpPipeline;
 import com.azure.core.http.HttpPipelineCallContext;
 import com.azure.core.http.HttpPipelineNextPolicy;
+import com.azure.core.http.HttpPipelineNextSyncPolicy;
 import com.azure.core.http.HttpPipelinePosition;
 import com.azure.core.http.HttpResponse;
+import com.azure.core.implementation.http.HttpPipelineNextSyncPolicyHelper;
 import reactor.core.publisher.Mono;
 
 /**
@@ -26,6 +28,17 @@ public interface HttpPipelinePolicy {
      */
     Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineNextPolicy next);
 
+    /**
+     * Processes provided request context and invokes the next policy synchronously.
+     *
+     * @param context The request context.
+     * @param next The next policy to invoke.
+     * @return A publisher that initiates the request upon subscription and emits a response on completion.
+     */
+    default HttpResponse processSync(HttpPipelineCallContext context, HttpPipelineNextSyncPolicy next) {
+        return process(context, HttpPipelineNextSyncPolicyHelper.toAsyncPolicy(next)).block();
+    }
+
     /**
      * Gets the position to place the policy.
      * <p>

sdk/core/azure-core/src/main/java/com/azure/core/http/policy/HttpPipelineSyncPolicy.java

@@ -0,0 +1,59 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.http.policy;
+
+import com.azure.core.http.HttpPipelineCallContext;
+import com.azure.core.http.HttpPipelineNextPolicy;
+import com.azure.core.http.HttpPipelineNextSyncPolicy;
+import com.azure.core.http.HttpResponse;
+import reactor.core.publisher.Mono;
+
+/**
+ * Represents a {@link HttpPipelinePolicy} that doesn't do any asynchronous or synchronously blocking operations.
+ */
+public class HttpPipelineSyncPolicy implements HttpPipelinePolicy {
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public final Mono<HttpResponse> process(HttpPipelineCallContext context, HttpPipelineNextPolicy next) {
+        return Mono.fromCallable(
+                () -> {
+                    beforeSendingRequest(context);
+                    return next;
+                })
+            .flatMap(ignored -> next.process())
+            .map(response -> afterReceivedResponse(context, response));
+    }
+
+    /**
+     * {@inheritDoc}
+     */
+    @Override
+    public final HttpResponse processSync(HttpPipelineCallContext context, HttpPipelineNextSyncPolicy next) {
+        beforeSendingRequest(context);
+        HttpResponse response = next.processSync();
+        return afterReceivedResponse(context, response);
+    }
+
+    /**
+     * Method is invoked before the request is sent.
+     * @param context The request context.
+     */
+    protected void beforeSendingRequest(HttpPipelineCallContext context) {
+        // empty by default
+    }
+
+    /**
+     * Method is invoked after the response is received.
+     * @param context The request context.
+     * @param response The response received.
+     * @return The transformed response.
+     */
+    protected HttpResponse afterReceivedResponse(HttpPipelineCallContext context, HttpResponse response) {
+        // empty by default
+        return response;
+    }
+}