Ensure that in-flight reactive result streams are DISCARDed

Reactive shall discard records on commit and/or rollback.
If summary has not yet arrived, tx.commit/rollback -> DISCARD n=-1, COMMIT/ROLL_BACK

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/internal/cursor/RxStatementResultCursor.java

@@ -40,7 +40,7 @@ public class RxStatementResultCursor implements Subscription, FailableCursor
     private final BasicPullResponseHandler pullHandler;
     private final Throwable runResponseError;
     private final CompletableFuture<ResultSummary> summaryFuture = new CompletableFuture<>();
-    boolean isRecordHandlerInstalled = false;
+    private boolean isRecordHandlerInstalled = false;
 
     public RxStatementResultCursor( RunResponseHandler runHandler, BasicPullResponseHandler pullHandler )
     {

driver/src/main/java/org/neo4j/driver/reactive/RxStatementResult.java

@@ -43,7 +43,7 @@ public interface RxStatementResult
     /**
      * Returns a cold publisher of keys.
      * <p>
-     * When this publisher is {@linkplain Publisher#subscribe(Subscriber) subscribed}, the query statement is sent to the server and get executed.
+     * When this publisher is {@linkplain Publisher#subscribe(Subscriber) subscribed}, the query statement is sent to the server and executed.
      * This method does not start the record streaming nor publish query execution error.
      * To retrieve the execution result, either {@link #records()} or {@link #summary()} can be used.
      * {@link #records()} starts record streaming and reports query execution error.

driver/src/test/java/org/neo4j/driver/integration/reactive/RxTransactionIT.java

@@ -177,6 +177,41 @@ void shouldBePossibleToRunMultipleStatementsWithoutWaiting( boolean commit )
         verifyCommittedOrRolledBack( commit );
     }
 
+    @ParameterizedTest
+    @MethodSource( "commit" )
+    void shouldRunStatementsOnResultPublish( boolean commit )
+    {
+        RxTransaction tx = await( Mono.from( session.beginTransaction() ) );
+
+        RxStatementResult cursor1 = tx.run( "CREATE (n:Person {name: 'Alice'}) RETURN n.name" );
+        RxStatementResult cursor2 = tx.run( "CREATE (n:Person {name: 'Bob'}) RETURN n.name" );
+
+        // The execution order is the same as the record publishing order.
+        List<Record> records = await( Flux.from( cursor2.records() ).concatWith( cursor1.records() ) );
+        assertThat( records.size(), equalTo( 2 ) );
+        assertThat( records.get( 0 ).get( "n.name" ).asString(), equalTo( "Bob" ) );
+        assertThat( records.get( 1 ).get( "n.name" ).asString(), equalTo( "Alice" ) );
+
+        assertCanCommitOrRollback( commit, tx );
+    }
+
+    @ParameterizedTest
+    @MethodSource( "commit" )
+    void shouldDiscardOnCommitOrRollback( boolean commit )
+    {
+        RxTransaction tx = await( Mono.from( session.beginTransaction() ) );
+        RxStatementResult cursor = tx.run( "UNWIND [1,2,3,4] AS a RETURN a" );
+
+        // We only perform run without any pull
+        await( Flux.from( cursor.keys() ) );
+        // We shall perform a discard here and then commit/rollback
+        assertCanCommitOrRollback( commit, tx );
+
+        // As a result the records size shall be 0.
+        List<Record> records = await( Flux.from( cursor.records() ) );
+        assertThat( records.size(), equalTo( 0 ) );
+    }
+
     @ParameterizedTest
     @MethodSource( "commit" )
     void shouldBePossibleToRunMultipleStatementsWithoutStreaming( boolean commit )
@@ -207,7 +242,6 @@ void shouldAllowRollbackAfterSingleWrongStatement()
         RxTransaction tx = await( Mono.from( session.beginTransaction() ) );
         assertFailToRunWrongStatement( tx );
         assertCanRollback( tx );
-
     }
 
     @Test

driver/src/test/java/org/neo4j/driver/internal/DirectDriverBoltKitTest.java

@@ -21,6 +21,8 @@
 import io.netty.channel.Channel;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+import reactor.core.publisher.Flux;
+import reactor.test.StepVerifier;
 
 import java.net.URI;
 import java.util.List;
@@ -43,6 +45,7 @@
 import org.neo4j.driver.internal.retry.RetrySettings;
 import org.neo4j.driver.internal.util.Clock;
 import org.neo4j.driver.internal.util.io.ChannelTrackingDriverFactory;
+import org.neo4j.driver.reactive.RxSession;
 import org.neo4j.driver.util.StubServer;
 
 import static java.util.Arrays.asList;
@@ -334,6 +337,25 @@ void shouldAllowDatabaseNameInBeginTransaction() throws Throwable
         }
     }
 
+    @Test
+    void shouldDiscardIfPullNotFinished() throws Throwable
+    {
+        StubServer server = StubServer.start( "read_tx_v4_discard.script", 9001 );
+
+        try ( Driver driver = GraphDatabase.driver( "bolt://localhost:9001", INSECURE_CONFIG ) )
+        {
+            Flux<String> keys = Flux.using(
+                    driver::rxSession,
+                    session -> session.readTransaction( tx -> tx.run( "UNWIND [1,2,3,4] AS a RETURN a" ).keys() ),
+                    RxSession::close );
+            StepVerifier.create( keys ).expectNext( "a" ).verifyComplete();
+        }
+        finally
+        {
+            assertEquals( 0, server.exitStatus() );
+        }
+    }
+
     private static void testTxCloseErrorPropagation( String script, Consumer<Transaction> txAction, String expectedErrorMessage )
             throws Exception
     {

driver/src/test/resources/read_tx_v4_discard.script

@@ -0,0 +1,13 @@
+!: BOLT 4
+!: AUTO RESET
+!: AUTO HELLO
+!: AUTO GOODBYE
+
+C: BEGIN { "mode": "r" }
+   RUN "UNWIND [1,2,3,4] AS a RETURN a" {} {}
+S: SUCCESS {}
+   SUCCESS {"t_first": 110, "fields": ["a"], "qid": 0}
+C: DISCARD {"qid": 0, "n": -1}
+S: SUCCESS {"type": "r", "t_last": 3, "db": "neo4j"}
+C: COMMIT
+S: SUCCESS {"bookmark": "e57085e2-727f-43f3-b632-7ec57978806e:117"}