deprecated retryWhen usage substituted (Azure#17658)

Co-authored-by: Sergey Novikov <[email protected]>
jaccx · Nov 30, 2020 · 750e80b · 750e80b
1 parent 62e60e5
commit 750e80b
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 4 deletions.
diff --git a/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RetryUtil.java b/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/RetryUtil.java
@@ -12,6 +12,7 @@
 import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.retry.Retry;
 
 import java.time.Duration;
 import java.util.Locale;
@@ -55,7 +56,7 @@ public static AmqpRetryPolicy getRetryPolicy(AmqpRetryOptions options) {
      */
     public static <T> Flux<T> withRetry(Flux<T> source, Duration operationTimeout, AmqpRetryPolicy retryPolicy) {
         return Flux.defer(() -> source.timeout(operationTimeout))
-            .retryWhen(errors -> retry(errors, retryPolicy));
+            .retryWhen(Retry.withThrowable(errors -> retry(errors, retryPolicy)));
     }
 
     /**
@@ -67,7 +68,7 @@ public static <T> Flux<T> withRetry(Flux<T> source, Duration operationTimeout, A
      */
     public static <T> Mono<T> withRetry(Mono<T> source, Duration operationTimeout, AmqpRetryPolicy retryPolicy) {
         return Mono.defer(() -> source.timeout(operationTimeout))
-            .retryWhen(errors -> retry(errors, retryPolicy));
+            .retryWhen(Retry.withThrowable(errors -> retry(errors, retryPolicy)));
     }
 
     private static Flux<Long> retry(Flux<Throwable> source, AmqpRetryPolicy retryPolicy) {

diff --git a/...osmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java b/...osmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java
@@ -5,6 +5,7 @@
 import com.azure.cosmos.implementation.directconnectivity.AddressSelector;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.util.retry.Retry;
 
 import java.time.Duration;
 import java.util.concurrent.Callable;
@@ -37,7 +38,7 @@ static public <T> Mono<T> executeRetry(Callable<Mono<T>> callbackMethod,
             } catch (Exception e) {
                 return Mono.error(e);
             }
-        }).retryWhen(RetryUtils.toRetryWhenFunc(retryPolicy));
+        }).retryWhen(Retry.withThrowable(RetryUtils.toRetryWhenFunc(retryPolicy)));
     }
 
     static public <T> Flux<T> fluxExecuteRetry(Callable<Flux<T>> callbackMethod, IRetryPolicy retryPolicy) {
@@ -48,7 +49,7 @@ static public <T> Flux<T> fluxExecuteRetry(Callable<Flux<T>> callbackMethod, IRe
             } catch (Exception e) {
                 return Flux.error(e);
             }
-        }).retryWhen(RetryUtils.toRetryWhenFunc(retryPolicy));
+        }).retryWhen(Retry.withThrowable(RetryUtils.toRetryWhenFunc(retryPolicy)));
     }
 
     static public <T> Mono<T> executeAsync(