Ensure error is propagated from rx queries correctly.

Zhen Li · Zhen Li · commit 438de66b2d93 · 2020-04-08T17:24:36.000+02:00
The rx queries should be semantically the same as the async queries.
Fixed a few wrong handling of finish future and error propergation used in rx queries.
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxFailingQuery.java b/driver/src/test/java/org/neo4j/driver/stress/RxFailingQuery.java
@@ -23,6 +23,7 @@
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.neo4j.driver.AccessMode;
 import org.neo4j.driver.Driver;
@@ -51,8 +52,7 @@ public CompletionStage<Void> execute( C context )
                 RxSession::close )
                 .subscribe( record -> {
                     assertThat( record.get( 0 ).asInt(), either( equalTo( 1 ) ).or( equalTo( 2 ) ) );
-                    queryFinished.complete( null );
-                }, error -> {
+                    }, error -> {
                     Throwable cause = Futures.completionExceptionCause( error );
                     assertThat( cause, is( arithmeticError() ) );
                     queryFinished.complete( null );
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxFailingQueryInTx.java b/driver/src/test/java/org/neo4j/driver/stress/RxFailingQueryInTx.java
@@ -52,7 +52,6 @@ public CompletionStage<Void> execute( C context )
                 RxTransaction::commit, ( tx, error ) -> tx.rollback(), null )
                 .subscribe( record -> {
                     assertThat( record.get( 0 ).asInt(), either( equalTo( 1 ) ).or( equalTo( 2 ) ) );
-                    queryFinished.complete( null );
                 }, error -> {
                     Throwable cause = Futures.completionExceptionCause( error );
                     assertThat( cause, is( arithmeticError() ) );
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxFailingQueryWithRetries.java b/driver/src/test/java/org/neo4j/driver/stress/RxFailingQueryWithRetries.java
@@ -51,7 +51,6 @@ public CompletionStage<Void> execute( C context )
                 RxSession::close )
                 .subscribe( record -> {
                     assertThat( record.get( 0 ).asInt(), either( equalTo( 1 ) ).or( equalTo( 2 ) ) );
-                    queryFinished.complete( null );
                 }, error -> {
                     Throwable cause = Futures.completionExceptionCause( error );
                     assertThat( cause, is( arithmeticError() ) );
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxWriteQuery.java b/driver/src/test/java/org/neo4j/driver/stress/RxWriteQuery.java
@@ -49,22 +49,23 @@ public CompletionStage<Void> execute( C context )
         Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.WRITE, context ) ),
                 session -> session.run( "CREATE ()" ).consume(), RxSession::close )
                 .subscribe( summary -> {
-                    queryFinished.complete( null );
                     assertEquals( 1, summary.counters().nodesCreated() );
                     context.nodeCreated();
-                }, error -> {
                     queryFinished.complete( null );
-                    handleError( Futures.completionExceptionCause( error ), context );
-                } );
+                }, error -> handleError( Futures.completionExceptionCause( error ), context, queryFinished ) );
 
         return queryFinished;
     }
 
-    private void handleError( Throwable error, C context )
+    private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )
     {
         if ( !stressTest.handleWriteFailure( error, context ) )
         {
-            throw new RuntimeException( error );
+            queryFinished.completeExceptionally( error );
+        }
+        else
+        {
+            queryFinished.complete( null );
         }
     }
 }
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxWriteQueryInTx.java b/driver/src/test/java/org/neo4j/driver/stress/RxWriteQueryInTx.java
@@ -53,18 +53,21 @@ public CompletionStage<Void> execute( C context )
                     context.nodeCreated();
                     queryFinished.complete( null );
                 }, error -> {
-                    handleError( Futures.completionExceptionCause( error ), context );
-                    queryFinished.complete( null );
+                    handleError( Futures.completionExceptionCause( error ), context, queryFinished );
                 } );
 
         return queryFinished;
     }
 
-    private void handleError( Throwable error, C context )
+    private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )
     {
         if ( !stressTest.handleWriteFailure( error, context ) )
         {
-            throw new RuntimeException( error );
+            queryFinished.completeExceptionally( error );
+        }
+        else
+        {
+            queryFinished.complete( null );
         }
     }
 }
diff --git a/driver/src/test/java/org/neo4j/driver/stress/RxWriteQueryWithRetries.java b/driver/src/test/java/org/neo4j/driver/stress/RxWriteQueryWithRetries.java
@@ -50,22 +50,23 @@ public CompletionStage<Void> execute( C context )
         Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.WRITE, context ) ),
                 session -> session.writeTransaction( tx -> tx.run( "CREATE ()" ).consume() ), RxSession::close )
                 .subscribe( summary -> {
-                    queryFinished.complete( null );
                     assertEquals( 1, summary.counters().nodesCreated() );
                     context.nodeCreated();
-                }, error -> {
                     queryFinished.complete( null );
-                    handleError( Futures.completionExceptionCause( error ), context );
-                } );
+                }, error -> handleError( Futures.completionExceptionCause( error ), context, queryFinished ) );
 
         return queryFinished;
     }
 
-    private void handleError( Throwable error, C context )
+    private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )
     {
         if ( !stressTest.handleWriteFailure( error, context ) )
         {
-            throw new RuntimeException( error );
+            queryFinished.completeExceptionally( error );
+        }
+        else
+        {
+            queryFinished.complete( null );
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -49,22 +49,23 @@ public CompletionStage<Void> execute( C context )`
`49`	`49`	`Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.WRITE, context ) ),`
`50`	`50`	`session -> session.run( "CREATE ()" ).consume(), RxSession::close )`
`51`	`51`	`.subscribe( summary -> {`
`52`		`- queryFinished.complete( null );`
`53`	`52`	`assertEquals( 1, summary.counters().nodesCreated() );`
`54`	`53`	`context.nodeCreated();`
`55`		`- }, error -> {`
`56`	`54`	`queryFinished.complete( null );`
`57`		`- handleError( Futures.completionExceptionCause( error ), context );`
`58`		`- } );`
	`55`	`+ }, error -> handleError( Futures.completionExceptionCause( error ), context, queryFinished ) );`
`59`	`56`
`60`	`57`	`return queryFinished;`
`61`	`58`	`}`
`62`	`59`
`63`		`- private void handleError( Throwable error, C context )`
	`60`	`+ private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )`
`64`	`61`	`{`
`65`	`62`	`if ( !stressTest.handleWriteFailure( error, context ) )`
`66`	`63`	`{`
`67`		`- throw new RuntimeException( error );`
	`64`	`+ queryFinished.completeExceptionally( error );`
	`65`	`+ }`
	`66`	`+ else`
	`67`	`+ {`
	`68`	`+ queryFinished.complete( null );`
`68`	`69`	`}`
`69`	`70`	`}`
`70`	`71`	`}`
Original file line number	Diff line number	Diff line change
`@@ -53,18 +53,21 @@ public CompletionStage<Void> execute( C context )`
`53`	`53`	`context.nodeCreated();`
`54`	`54`	`queryFinished.complete( null );`
`55`	`55`	`}, error -> {`
`56`		`- handleError( Futures.completionExceptionCause( error ), context );`
`57`		`- queryFinished.complete( null );`
	`56`	`+ handleError( Futures.completionExceptionCause( error ), context, queryFinished );`
`58`	`57`	`} );`
`59`	`58`
`60`	`59`	`return queryFinished;`
`61`	`60`	`}`
`62`	`61`
`63`		`- private void handleError( Throwable error, C context )`
	`62`	`+ private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )`
`64`	`63`	`{`
`65`	`64`	`if ( !stressTest.handleWriteFailure( error, context ) )`
`66`	`65`	`{`
`67`		`- throw new RuntimeException( error );`
	`66`	`+ queryFinished.completeExceptionally( error );`
	`67`	`+ }`
	`68`	`+ else`
	`69`	`+ {`
	`70`	`+ queryFinished.complete( null );`
`68`	`71`	`}`
`69`	`72`	`}`
`70`	`73`	`}`
Original file line number	Diff line number	Diff line change
`@@ -50,22 +50,23 @@ public CompletionStage<Void> execute( C context )`
`50`	`50`	`Flux.usingWhen( Mono.fromSupplier( () -> newSession( AccessMode.WRITE, context ) ),`
`51`	`51`	`session -> session.writeTransaction( tx -> tx.run( "CREATE ()" ).consume() ), RxSession::close )`
`52`	`52`	`.subscribe( summary -> {`
`53`		`- queryFinished.complete( null );`
`54`	`53`	`assertEquals( 1, summary.counters().nodesCreated() );`
`55`	`54`	`context.nodeCreated();`
`56`		`- }, error -> {`
`57`	`55`	`queryFinished.complete( null );`
`58`		`- handleError( Futures.completionExceptionCause( error ), context );`
`59`		`- } );`
	`56`	`+ }, error -> handleError( Futures.completionExceptionCause( error ), context, queryFinished ) );`
`60`	`57`
`61`	`58`	`return queryFinished;`
`62`	`59`	`}`
`63`	`60`
`64`		`- private void handleError( Throwable error, C context )`
	`61`	`+ private void handleError( Throwable error, C context, CompletableFuture<Void> queryFinished )`
`65`	`62`	`{`
`66`	`63`	`if ( !stressTest.handleWriteFailure( error, context ) )`
`67`	`64`	`{`
`68`		`- throw new RuntimeException( error );`
	`65`	`+ queryFinished.completeExceptionally( error );`
	`66`	`+ }`
	`67`	`+ else`
	`68`	`+ {`
	`69`	`+ queryFinished.complete( null );`
`69`	`70`	`}`
`70`	`71`	`}`
`71`	`72`	`}`