Using new windowTimeout operator with backpressure support

eng/versioning/version_client.txt

@@ -354,7 +354,8 @@ com.azure.resourcemanager:azure-resourcemanager-education;1.0.0-beta.1;1.0.0-bet
 com.azure.resourcemanager:azure-resourcemanager-orbital;1.0.0-beta.1;1.0.0-beta.2
 com.azure.tools:azure-sdk-archetype;1.0.0;1.2.0-beta.1
 com.azure.tools:azure-sdk-build-tool;1.0.0-beta.1;1.0.0-beta.2
-
+unreleased_com.azure:azure-core-amqp;2.6.0-beta.1
+unreleased_com.azure:azure-core;1.30.0-beta.1
 
 # Unreleased dependencies: Copy the entry from above, prepend "unreleased_" and remove the current
 # version. Unreleased dependencies are only valid for dependency versions.

sdk/eventhubs/azure-messaging-eventhubs/pom.xml

@@ -37,12 +37,12 @@
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core</artifactId>
-      <version>1.29.1</version> <!-- {x-version-update;com.azure:azure-core;dependency} -->
+      <version>1.30.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
     </dependency>
     <dependency>
       <groupId>com.azure</groupId>
       <artifactId>azure-core-amqp</artifactId>
-      <version>2.5.2</version> <!-- {x-version-update;com.azure:azure-core-amqp;dependency} -->
+      <version>2.6.0-beta.1</version> <!-- {x-version-update;unreleased_com.azure:azure-core-amqp;dependency} -->
     </dependency>
 
     <!-- Test dependencies -->

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/PartitionPumpManager.java

@@ -10,6 +10,7 @@
 import com.azure.core.util.tracing.ProcessKind;
 import com.azure.messaging.eventhubs.implementation.PartitionProcessor;
 import com.azure.messaging.eventhubs.implementation.PartitionProcessorException;
+import com.azure.messaging.eventhubs.implementation.ReactorShim;
 import com.azure.messaging.eventhubs.models.Checkpoint;
 import com.azure.messaging.eventhubs.models.CloseContext;
 import com.azure.messaging.eventhubs.models.CloseReason;
@@ -244,8 +245,7 @@ void startPartitionPump(PartitionOwnership claimedOwnership, Checkpoint checkpoi
                 });
 
             if (maxWaitTime != null) {
-                partitionEventFlux = receiver
-                    .windowTimeout(maxBatchSize, maxWaitTime);
+                partitionEventFlux = ReactorShim.windowTimeout(receiver, maxBatchSize, maxWaitTime);
             } else {
                 partitionEventFlux = receiver
                     .window(maxBatchSize);

sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/implementation/ReactorShim.java

@@ -0,0 +1,87 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.messaging.eventhubs.implementation;
+
+import com.azure.core.util.logging.ClientLogger;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Schedulers;
+
+import java.lang.invoke.MethodHandle;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.MethodType;
+import java.time.Duration;
+
+/**
+ * A shim that proxies Reactor operator calls, for example, if the loaded Reactor has an optimized variant
+ * of a standard operator, then shim uses it else fallback to the standard variant, if there are breaking
+ * changes in operators among Reactor versions that SDK supports, then shim may expose a unified API for
+ * those operators.
+ */
+public final class ReactorShim {
+    private static final ClientLogger LOGGER = new ClientLogger(ReactorShim.class);
+
+    /* Reactor Operator names */
+    private static final String  WINDOW_TIMEOUT_OPERATOR = "windowTimeout";
+    /* Reactor Operator handles */
+    private static final MethodHandle BACKPRESSURE_WINDOW_TIMEOUT_OPERATOR_HANDLE;
+
+    static {
+        BACKPRESSURE_WINDOW_TIMEOUT_OPERATOR_HANDLE = lookupBackpressureWindowTimeoutOperator();
+    }
+
+    /**
+     * Split the {@code source} {@link Flux} sequence into multiple {@link Flux} windows containing
+     * {@code maxSize} elements (or less for the final window) and starting from the first item.
+     * Each {@link Flux} window will onComplete once it contains {@code maxSize} elements
+     * OR it has been open for the given {@link Duration} (as measured on the {@link Schedulers#parallel() parallel}
+     * Scheduler).
+     *
+     * <p>
+     * If the loaded Reactor library has a backpressure-aware window-timeout operator then it will be used,
+     * which caps requests to the source by {@code maxSize} (i.e. prefetch), otherwise, the regular variant
+     * of window-timeout operator requesting unbounded demand will be used.
+     *
+     * @param maxSize the maximum number of items to emit in the window before closing it
+     * @param maxTime the maximum {@link Duration} since the window was opened before closing it
+     *
+     * @param <T> the element type of the source {@link Flux}.
+     * @return a {@link Flux} of {@link Flux} windows based on element count and duration.
+     */
+    public static <T> Flux<Flux<T>> windowTimeout(Flux<T> source, int maxSize, Duration maxTime) {
+        if (BACKPRESSURE_WINDOW_TIMEOUT_OPERATOR_HANDLE == null) {
+            // optimized (backpressure) aware windowTimeout operator not available use standard variant.
+            return source.windowTimeout(maxSize, maxTime);
+        }
+
+        try {
+            return ((Flux<Flux<T>>) (BACKPRESSURE_WINDOW_TIMEOUT_OPERATOR_HANDLE.invoke(source, maxSize, maxTime, true)));
+        } catch (Throwable err) {
+            // 'java.lang.invoke' throws Throwable. Given 'Error' category represents a serious
+            // abnormal thread state throw it immediately else throw via standard azure-core Logger.
+            if (err instanceof Error) {
+                throw (Error) err;
+            } else if (err instanceof RuntimeException) {
+                throw LOGGER.logExceptionAsError((RuntimeException) err);
+            } else {
+                throw LOGGER.logExceptionAsError(new RuntimeException(err));
+            }
+        }
+    }
+
+    /**
+     * Try to obtain {@link MethodHandle} for backpressure aware windowTimeout Reactor operator.
+     *
+     * @return if the backpressure aware windowTimeout Reactor operator is available then return
+     * operator {@link MethodHandle} else null.
+     */
+    private static MethodHandle lookupBackpressureWindowTimeoutOperator() {
+        try {
+            return MethodHandles.publicLookup().findVirtual(Flux.class, WINDOW_TIMEOUT_OPERATOR,
+                MethodType.methodType(Flux.class, int.class, Duration.class, boolean.class));
+        } catch (IllegalAccessException | NoSuchMethodException err) {
+            LOGGER.verbose("Failed to retrieve MethodHandle for backpressure aware windowTimeout Reactor operator.", err);
+        }
+        return null;
+    }
+}