Remove reactor-bom and upgrade to reactor 3.4.7

Author: injectives

driver/src/main/java/org/neo4j/driver/internal/retry/ExponentialBackoffRetryLogic.java

@@ -25,16 +25,15 @@
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
 import reactor.util.context.Context;
-import reactor.util.function.Tuples;
+import reactor.util.retry.Retry;
 
 import java.time.Duration;
-import java.util.ArrayList;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
 import java.util.function.Supplier;
 
 import org.neo4j.driver.Logger;
@@ -145,7 +144,7 @@ public <T> CompletionStage<T> retryAsync( Supplier<CompletionStage<T>> work )
     @Override
     public <T> Publisher<T> retryRx( Publisher<T> work )
     {
-        return Flux.from( work ).retryWhen( retryRxCondition() );
+        return Flux.from( work ).retryWhen( exponentialBackoffRetryRx() );
     }
 
     protected boolean canRetryOn( Throwable error )
@@ -177,48 +176,50 @@ private static Throwable extractPossibleTerminationCause( Throwable error )
         return error;
     }
 
-    private Function<Flux<Throwable>,Publisher<Context>> retryRxCondition()
+    private Retry exponentialBackoffRetryRx()
     {
-        return errorCurrentAttempt -> errorCurrentAttempt.flatMap( e -> Mono.subscriberContext().map( ctx -> Tuples.of( e, ctx ) ) ).flatMap( t2 ->
-        {
-
-            Throwable throwable = t2.getT1();
-            Throwable error = extractPossibleTerminationCause( throwable );
-
-            Context ctx = t2.getT2();
-
-            List<Throwable> errors = ctx.getOrDefault( "errors", null );
-
-            long startTime = ctx.getOrDefault( "startTime", -1L );
-            long nextDelayMs = ctx.getOrDefault( "nextDelayMs", initialRetryDelayMs );
-
-            if ( canRetryOn( error ) )
-            {
-                long currentTime = clock.millis();
-                if ( startTime == -1 )
+        return Retry.from( retrySignals -> retrySignals.transformDeferredContextual( ( flux, contextView ) -> flux.map(
+                signal ->
                 {
-                    startTime = currentTime;
-                }
+                    Throwable throwable = signal.failure();
+                    Throwable error = extractPossibleTerminationCause( throwable );
 
-                long elapsedTime = currentTime - startTime;
-                if ( elapsedTime < maxRetryTimeMs )
-                {
-                    long delayWithJitterMs = computeDelayWithJitter( nextDelayMs );
-                    log.warn( "Reactive transaction failed and is scheduled to retry in " + delayWithJitterMs + "ms", error );
+                    List<Throwable> errors = contextView.getOrDefault( "errors", null );
 
-                    nextDelayMs = (long) (nextDelayMs * multiplier);
-                    errors = recordError( error, errors );
+                    long startTime = contextView.getOrDefault( "startTime", -1L );
+                    long nextDelayMs = contextView.getOrDefault( "nextDelayMs", initialRetryDelayMs );
 
-                    // retry on netty event loop thread
-                    EventExecutor eventExecutor = eventExecutorGroup.next();
-                    return Mono.just( ctx.put( "errors", errors ).put( "startTime", startTime ).put( "nextDelayMs", nextDelayMs ) ).delayElement(
-                            Duration.ofMillis( delayWithJitterMs ), Schedulers.fromExecutorService( eventExecutor ) );
-                }
-            }
-            addSuppressed( throwable, errors );
+                    if ( canRetryOn( error ) )
+                    {
+                        long currentTime = clock.millis();
+                        if ( startTime == -1 )
+                        {
+                            startTime = currentTime;
+                        }
+
+                        long elapsedTime = currentTime - startTime;
+                        if ( elapsedTime < maxRetryTimeMs )
+                        {
+                            long delayWithJitterMs = computeDelayWithJitter( nextDelayMs );
+                            log.warn( "Reactive transaction failed and is scheduled to retry in " + delayWithJitterMs + "ms", error );
+
+                            nextDelayMs = (long) (nextDelayMs * multiplier);
+                            errors = recordError( error, errors );
+
+                            // retry on netty event loop thread
+                            EventExecutor eventExecutor = eventExecutorGroup.next();
+                            Context context = Context.of(
+                                    "errors", errors,
+                                    "startTime", startTime,
+                                    "nextDelayMs", nextDelayMs
+                            );
+                            return Mono.just( context ).delayElement( Duration.ofMillis( delayWithJitterMs ), Schedulers.fromExecutorService( eventExecutor ) );
+                        }
+                    }
+                    addSuppressed( throwable, errors );
 
-            return Mono.error( throwable );
-        } );
+                    return Mono.error( throwable );
+                } ) ) );
     }
 
     private <T> void executeWorkInEventLoop( CompletableFuture<T> resultFuture, Supplier<CompletionStage<T>> work )
@@ -373,7 +374,7 @@ private static List<Throwable> recordError( Throwable error, List<Throwable> err
     {
         if ( errors == null )
         {
-            errors = new ArrayList<>();
+            errors = new LinkedList<>();
         }
         errors.add( error );
         return errors;

pom.xml

@@ -32,7 +32,7 @@
     <!-- Please note that when updating this dependency -->
     <!-- (i.e. due to a security vulnerability or bug) that the -->
     <!-- corresponding server dependency also needs updating.-->
-    <reactor-bom.version>Dysprosium-SR21</reactor-bom.version>
+    <reactor.version>3.4.7</reactor.version>
     <rxjava.version>2.2.21</rxjava.version>
     <slf4j-api.version>1.7.31</slf4j-api.version>
     <hamcrest-junit.version>2.0.0.0</hamcrest-junit.version>
@@ -90,10 +90,8 @@
       </dependency>
       <dependency>
         <groupId>io.projectreactor</groupId>
-        <artifactId>reactor-bom</artifactId>
-        <version>${reactor-bom.version}</version>
-        <type>pom</type>
-        <scope>import</scope>
+        <artifactId>reactor-core</artifactId>
+        <version>${reactor.version}</version>
       </dependency>
 
       <!--Compile dependencies only used by Examples-->
@@ -111,6 +109,12 @@
       </dependency>
 
       <!-- Test dependencies -->
+      <dependency>
+        <groupId>io.projectreactor</groupId>
+        <artifactId>reactor-test</artifactId>
+        <version>${reactor.version}</version>
+        <scope>test</scope>
+      </dependency>
       <dependency>
         <groupId>org.hamcrest</groupId>
         <artifactId>hamcrest-junit</artifactId>