Performance Improvements for FluxUtils

sdk/core/azure-core/src/main/java/com/azure/core/http/rest/ResponseConstructorsCache.java

@@ -10,58 +10,66 @@
 import com.azure.core.util.logging.ClientLogger;
 import reactor.core.publisher.Mono;
 
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
 import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
 import java.util.Arrays;
 import java.util.Comparator;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
-import java.util.function.Supplier;
 
 /**
  * A concurrent cache of {@link Response} constructors.
  */
 final class ResponseConstructorsCache {
+    private static final String THREE_PARAM_EXCEPTION = "Failed to deserialize 3-parameter response.";
+    private static final String FOUR_PARAM_EXCEPTION = "Failed to deserialize 4-parameter response.";
+    private static final String FIVE_PARAM_EXCEPTION_HEADERS = "Failed to deserialize 5-parameter response with "
+        + "decoded headers.";
+    private static final String FIVE_PARAM_EXCEPTION_NO_HEADERS = "Failed to deserialize 5-parameter response "
+        + "without decoded headers.";
+    private static final String INVALID_PARAMETER_COUNT = "Response constructor with expected parameters not found.";
+
     private final ClientLogger logger = new ClientLogger(ResponseConstructorsCache.class);
-    private final Map<Class<?>, Constructor<? extends Response<?>>> cache = new ConcurrentHashMap<>();
+    private final Map<Class<?>, MethodHandle> cache = new ConcurrentHashMap<>();
 
     /**
      * Identify the suitable constructor for the given response class.
      *
      * @param responseClass the response class
      * @return identified constructor, null if there is no match
      */
-    Constructor<? extends Response<?>> get(Class<? extends Response<?>> responseClass) {
+    MethodHandle get(Class<? extends Response<?>> responseClass) {
         return this.cache.computeIfAbsent(responseClass, this::locateResponseConstructor);
     }
 
     /**
      * Identify the most specific constructor for the given response class.
-     *
+     * <p>
      * The most specific constructor is looked up following order:
-     * 1. (httpRequest, statusCode, headers, body, decodedHeaders)
-     * 2. (httpRequest, statusCode, headers, body)
-     * 3. (httpRequest, statusCode, headers)
+     * <ol>
+     * <li>(httpRequest, statusCode, headers, body, decodedHeaders)</li>
+     * <li>(httpRequest, statusCode, headers, body)</li>
+     * <li>(httpRequest, statusCode, headers)</li>
+     * </ol>
      *
-     * Developer Note: This method logic can be easily replaced with Java.Stream
-     * and associated operators but we're using basic sort and loop constructs
-     * here as this method is in hot path and Stream route is consuming a fair
+     * Developer Note: This method logic can be easily replaced with Java.Stream and associated operators but we're
+     * using basic sort and loop constructs here as this method is in hot path and Stream route is consuming a fair
      * amount of resources.
      *
      * @param responseClass the response class
      * @return identified constructor, null if there is no match
      */
-    @SuppressWarnings("unchecked")
-    private Constructor<? extends Response<?>> locateResponseConstructor(Class<?> responseClass) {
+    private MethodHandle locateResponseConstructor(Class<?> responseClass) {
         Constructor<?>[] constructors = responseClass.getDeclaredConstructors();
         // Sort constructors in the "descending order" of parameter count.
         Arrays.sort(constructors, Comparator.comparing(Constructor::getParameterCount, (a, b) -> b - a));
         for (Constructor<?> constructor : constructors) {
             final int paramCount = constructor.getParameterCount();
             if (paramCount >= 3 && paramCount <= 5) {
                 try {
-                    return (Constructor<? extends Response<?>>) constructor;
+                    return MethodHandles.lookup().unreflectConstructor(constructor);
                 } catch (Throwable t) {
                     throw logger.logExceptionAsError(new RuntimeException(t));
                 }
@@ -78,64 +86,49 @@ private Constructor<? extends Response<?>> locateResponseConstructor(Class<?> re
      * @param bodyAsObject the http response content
      * @return an instance of a {@link Response} implementation
      */
-    Mono<Response<?>> invoke(final Constructor<? extends Response<?>> constructor,
-                             final HttpResponseDecoder.HttpDecodedResponse decodedResponse,
-                             final Object bodyAsObject) {
+    Mono<Response<?>> invoke(final MethodHandle constructor,
+        final HttpResponseDecoder.HttpDecodedResponse decodedResponse, final Object bodyAsObject) {
         final HttpResponse httpResponse = decodedResponse.getSourceResponse();
         final HttpRequest httpRequest = httpResponse.getRequest();
         final int responseStatusCode = httpResponse.getStatusCode();
         final HttpHeaders responseHeaders = httpResponse.getHeaders();
 
-        final int paramCount = constructor.getParameterCount();
+        final int paramCount = constructor.type().parameterCount();
         switch (paramCount) {
             case 3:
                 try {
-                    return Mono.just(constructor.newInstance(httpRequest,
-                        responseStatusCode,
+                    return Mono.just((Response<?>) constructor.invoke(httpRequest, responseStatusCode,
                         responseHeaders));
-                } catch (IllegalAccessException | InvocationTargetException | InstantiationException e) {
-                    throw logger.logExceptionAsError(new RuntimeException("Failed to deserialize 3-parameter"
-                        + " response. ", e));
+                } catch (Throwable e) {
+                    throw logger.logExceptionAsError(new RuntimeException(THREE_PARAM_EXCEPTION, e));
                 }
             case 4:
                 try {
-                    return Mono.just(constructor.newInstance(httpRequest,
-                        responseStatusCode,
-                        responseHeaders,
+                    return Mono.just((Response<?>) constructor.invoke(httpRequest, responseStatusCode, responseHeaders,
                         bodyAsObject));
-                } catch (IllegalAccessException | InvocationTargetException | InstantiationException e) {
-                    throw logger.logExceptionAsError(new RuntimeException("Failed to deserialize 4-parameter"
-                        + " response. ", e));
+                } catch (Throwable e) {
+                    throw logger.logExceptionAsError(new RuntimeException(FOUR_PARAM_EXCEPTION, e));
                 }
             case 5:
                 return decodedResponse.getDecodedHeaders()
                     .map((Function<Object, Response<?>>) decodedHeaders -> {
                         try {
-                            return constructor.newInstance(httpRequest,
-                                responseStatusCode,
-                                responseHeaders,
-                                bodyAsObject,
-                                decodedHeaders);
-                        } catch (IllegalAccessException | InvocationTargetException | InstantiationException e) {
-                            throw logger.logExceptionAsError(new RuntimeException("Failed to deserialize 5-parameter"
-                                + " response with decoded headers. ", e));
+                            return (Response<?>) constructor.invoke(httpRequest, responseStatusCode, responseHeaders,
+                                bodyAsObject, decodedHeaders);
+                        } catch (Throwable e) {
+                            throw logger.logExceptionAsError(new RuntimeException(FIVE_PARAM_EXCEPTION_HEADERS, e));
                         }
                     })
-                    .switchIfEmpty(Mono.defer((Supplier<Mono<Response<?>>>) () -> {
+                    .switchIfEmpty(Mono.defer(() -> {
                         try {
-                            return Mono.just(constructor.newInstance(httpRequest,
-                                responseStatusCode,
-                                responseHeaders,
-                                bodyAsObject,
-                                null));
-                        } catch (IllegalAccessException | InvocationTargetException | InstantiationException e) {
-                            throw logger.logExceptionAsError(new RuntimeException(
-                                "Failed to deserialize 5-parameter response without decoded headers.", e));
+                            return Mono.just((Response<?>) constructor.invoke(httpRequest, responseStatusCode,
+                                responseHeaders, bodyAsObject, null));
+                        } catch (Throwable e) {
+                            throw logger.logExceptionAsError(new RuntimeException(FIVE_PARAM_EXCEPTION_NO_HEADERS, e));
                         }
                     }));
             default:
-                throw logger.logExceptionAsError(
-                    new IllegalStateException("Response constructor with expected parameters not found."));
+                throw logger.logExceptionAsError(new IllegalStateException(INVALID_PARAMETER_COUNT));
         }
     }
 }

sdk/core/azure-core/src/main/java/com/azure/core/http/rest/RestProxy.java

@@ -24,6 +24,7 @@
 import com.azure.core.implementation.serializer.HttpResponseDecoder.HttpDecodedResponse;
 import com.azure.core.util.Base64Url;
 import com.azure.core.util.Context;
+import com.azure.core.util.CoreUtils;
 import com.azure.core.util.FluxUtil;
 import com.azure.core.util.UrlBuilder;
 import com.azure.core.util.logging.ClientLogger;
@@ -38,6 +39,7 @@
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.lang.invoke.MethodHandle;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
@@ -71,6 +73,7 @@ public final class RestProxy implements InvocationHandler {
     private final SwaggerInterfaceParser interfaceParser;
     private final HttpResponseDecoder decoder;
 
+    // Should this be static?
     private final ResponseConstructorsCache responseConstructorsCache;
 
     /**
@@ -254,7 +257,7 @@ private HttpRequest configRequest(final HttpRequest request, final SwaggerMethod
 
             // If this is null or empty, the service interface definition is incomplete and should
             // be fixed to ensure correct definitions are applied
-            if (contentType == null || contentType.isEmpty()) {
+            if (CoreUtils.isNullOrEmpty(contentType)) {
                 if (bodyContentObject instanceof byte[] || bodyContentObject instanceof String) {
                     contentType = ContentType.APPLICATION_OCTET_STREAM;
                 } else {
@@ -315,9 +318,7 @@ private Mono<HttpDecodedResponse> ensureExpectedStatus(final Mono<HttpDecodedRes
     }
 
     private static Exception instantiateUnexpectedException(final UnexpectedExceptionInformation exception,
-        final HttpResponse httpResponse,
-        final byte[] responseContent,
-        final Object responseDecodedContent) {
+        final HttpResponse httpResponse, final byte[] responseContent, final Object responseDecodedContent) {
         final int responseStatusCode = httpResponse.getStatusCode();
         final String contentType = httpResponse.getHeaderValue("Content-Type");
         final String bodyRepresentation;
@@ -331,13 +332,18 @@ private static Exception instantiateUnexpectedException(final UnexpectedExceptio
 
         Exception result;
         try {
+            /*
+             * Could we cache this constructor?
+             */
             final Constructor<? extends HttpResponseException> exceptionConstructor =
                 exception.getExceptionType().getConstructor(String.class, HttpResponse.class,
                     exception.getExceptionBodyType());
             result = exceptionConstructor.newInstance("Status code " + responseStatusCode + ", " + bodyRepresentation,
-                httpResponse,
-                responseDecodedContent);
+                httpResponse, responseDecodedContent);
         } catch (ReflectiveOperationException e) {
+            /*
+             * Would this be worth using a StringBuilder?
+             */
             String message = "Status code " + responseStatusCode + ", but an instance of "
                 + exception.getExceptionType().getCanonicalName() + " cannot be created."
                 + " Response body: " + bodyRepresentation;
@@ -363,48 +369,35 @@ private Mono<HttpDecodedResponse> ensureExpectedStatus(final HttpDecodedResponse
         final SwaggerMethodParser methodParser) {
         final int responseStatusCode = decodedResponse.getSourceResponse().getStatusCode();
         final Mono<HttpDecodedResponse> asyncResult;
-        if (!methodParser.isExpectedResponseStatusCode(responseStatusCode)) {
-            Mono<byte[]> bodyAsBytes = decodedResponse.getSourceResponse().getBodyAsByteArray();
-
-            asyncResult = bodyAsBytes.flatMap((Function<byte[], Mono<HttpDecodedResponse>>) responseContent -> {
-                // bodyAsString() emits non-empty string, now look for decoded version of same string
-                Mono<Object> decodedErrorBody = decodedResponse.getDecodedBody(responseContent);
-
-                return decodedErrorBody
-                    .flatMap((Function<Object, Mono<HttpDecodedResponse>>) responseDecodedErrorObject -> {
-                        // decodedBody() emits 'responseDecodedErrorObject' the successfully decoded exception
-                        // body object
-                        Throwable exception =
-                            instantiateUnexpectedException(methodParser.getUnexpectedException(responseStatusCode),
-                                decodedResponse.getSourceResponse(),
-                                responseContent,
-                                responseDecodedErrorObject);
-                        return Mono.error(exception);
-                    })
-                    .switchIfEmpty(Mono.defer((Supplier<Mono<HttpDecodedResponse>>) () -> {
-                        // decodedBody() emits empty, indicate unable to decode 'responseContent',
-                        // create exception with un-decodable content string and without exception body object.
-                        Throwable exception =
-                            instantiateUnexpectedException(methodParser.getUnexpectedException(responseStatusCode),
-                                decodedResponse.getSourceResponse(),
-                                responseContent,
-                                null);
-                        return Mono.error(exception);
-                    }));
-            }).switchIfEmpty(Mono.defer((Supplier<Mono<HttpDecodedResponse>>) () -> {
-                // bodyAsString() emits empty, indicate no body, create exception empty content string no exception
-                // body object.
-                Throwable exception =
-                    instantiateUnexpectedException(methodParser.getUnexpectedException(responseStatusCode),
-                        decodedResponse.getSourceResponse(),
-                        null,
-                        null);
-                return Mono.error(exception);
-            }));
-        } else {
-            asyncResult = Mono.just(decodedResponse);
+
+        if (methodParser.isExpectedResponseStatusCode(responseStatusCode)) {
+            return Mono.just(decodedResponse);
         }
-        return asyncResult;
+
+        UnexpectedExceptionInformation exceptionInformation = methodParser.getUnexpectedException(responseStatusCode);
+        HttpResponse response = decodedResponse.getSourceResponse();
+
+        Mono<byte[]> bodyAsBytes = decodedResponse.getSourceResponse().getBodyAsByteArray();
+        return bodyAsBytes.flatMap((Function<byte[], Mono<HttpDecodedResponse>>) responseContent -> {
+            // bodyAsString() emits non-empty string, now look for decoded version of same string
+            return decodedResponse.getDecodedBody(responseContent)
+                /*
+                 * getDecodedBody() emits the successfully decoded exception body object.
+                 */
+                .flatMap((Function<Object, Mono<HttpDecodedResponse>>) responseDecodedErrorObject -> Mono.error(
+                    instantiateUnexpectedException(exceptionInformation, response, responseContent,
+                        responseDecodedErrorObject)))
+                .switchIfEmpty(Mono.defer((Supplier<Mono<HttpDecodedResponse>>) () -> {
+                    // decodedBody() emits empty, indicate unable to decode 'responseContent',
+                    // create exception with un-decodable content string and without exception body object.
+                    return Mono.error(instantiateUnexpectedException(exceptionInformation, response, responseContent,
+                        null));
+                }));
+        }).switchIfEmpty(Mono.defer((Supplier<Mono<HttpDecodedResponse>>) () -> {
+            // bodyAsString() emits empty, indicate no body, create exception empty content string no exception
+            // body object.
+            return Mono.error(instantiateUnexpectedException(exceptionInformation, response, null, null));
+        }));
     }
 
     private Mono<?> handleRestResponseReturnType(final HttpDecodedResponse response,
@@ -444,7 +437,7 @@ private Mono<Response<?>> createResponse(HttpDecodedResponse response, Type enti
             }
         }
 
-        Constructor<? extends Response<?>> ctr = this.responseConstructorsCache.get(cls);
+        MethodHandle ctr = this.responseConstructorsCache.get(cls);
         if (ctr != null) {
             return this.responseConstructorsCache.invoke(ctr, response, bodyAsObject);
         } else {
@@ -493,9 +486,7 @@ private Mono<?> handleBodyReturnType(final HttpDecodedResponse response,
      * @return the deserialized result
      */
     private Object handleRestReturnType(final Mono<HttpDecodedResponse> asyncHttpDecodedResponse,
-        final SwaggerMethodParser methodParser,
-        final Type returnType,
-        final Context context) {
+        final SwaggerMethodParser methodParser, final Type returnType, final Context context) {
         final Mono<HttpDecodedResponse> asyncExpectedResponse =
             ensureExpectedStatus(asyncHttpDecodedResponse, methodParser)
                 .doOnEach(RestProxy::endTracingSpan)
@@ -587,23 +578,10 @@ private static SerializerAdapter createDefaultSerializer() {
      * @return the default HttpPipeline
      */
     private static HttpPipeline createDefaultPipeline() {
-        return createDefaultPipeline(null);
-    }
-
-    /**
-     * Create the default HttpPipeline.
-     *
-     * @param credentialsPolicy the credentials policy factory to use to apply authentication to the pipeline
-     * @return the default HttpPipeline
-     */
-    private static HttpPipeline createDefaultPipeline(HttpPipelinePolicy credentialsPolicy) {
         List<HttpPipelinePolicy> policies = new ArrayList<>();
         policies.add(new UserAgentPolicy());
         policies.add(new RetryPolicy());
         policies.add(new CookiePolicy());
-        if (credentialsPolicy != null) {
-            policies.add(credentialsPolicy);
-        }
 
         return new HttpPipelineBuilder()
             .policies(policies.toArray(new HttpPipelinePolicy[0]))