Recreated the error

Author: Zhen Li

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

@@ -26,6 +26,7 @@
 import reactor.test.StepVerifier;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -38,8 +39,8 @@
 import org.neo4j.driver.exceptions.SessionExpiredException;
 import org.neo4j.driver.exceptions.TransientException;
 import org.neo4j.driver.internal.util.EnabledOnNeo4jWith;
-import org.neo4j.driver.reactive.RxStatementResult;
 import org.neo4j.driver.reactive.RxSession;
+import org.neo4j.driver.reactive.RxStatementResult;
 import org.neo4j.driver.reactive.RxTransaction;
 import org.neo4j.driver.reactive.RxTransactionWork;
 import org.neo4j.driver.util.DatabaseExtension;
@@ -182,6 +183,43 @@ void shouldRunAsyncTransactionThatCanNotBeRetriedAfterATransientFailure()
         assertNoParallelScheduler();
     }
 
+    @Test
+    void shouldHandleNestedQueries() {
+        int size = 12555;
+
+        Flux<Integer> nodeIds = Flux.using( neo4j.driver()::rxSession,
+                session -> Flux.from( session.run( "UNWIND range(1, $size) AS x RETURN x",
+                        Collections.singletonMap( "size", size ) ).records() )
+                        .limitRate( 20 )
+                        .flatMap( record -> {
+                            int x = record.get( "x" ).asInt();
+                            RxStatementResult innerResult = session.run( "CREATE (n:Node {id: $x}) RETURN n.id",
+                                    Collections.singletonMap( "x", x ) );
+                            return innerResult.records();
+                        } ).map( r -> r.get( 0 ).asInt() ), RxSession::close );
+
+        StepVerifier.create( nodeIds ).expectNextCount( size ).verifyComplete();
+    }
+
+    @Test
+    void shouldGiveHelpfulErrorMessageWhenNestingTransactionFunctions() {
+
+        int size = 12555;
+        Flux<Integer> nodeIds = Flux.using( neo4j.driver()::rxSession,
+                session -> Flux.from( session.readTransaction( tx ->
+                        tx.run( "UNWIND range(1, $size) AS x RETURN x",
+                        Collections.singletonMap( "size", size ) ).records() ) )
+                        .limitRate( 20 )
+                        .flatMap( record -> {
+                            int x = record.get( "x" ).asInt();
+                            return session.writeTransaction( tx ->
+                                    tx.run( "CREATE (n:Node {id: $x}) RETURN n.id",
+                                    Collections.singletonMap( "x", x ) ).records() );
+                        } ).map( r -> r.get( 0 ).asInt() ), RxSession::close );
+
+        StepVerifier.create( nodeIds ).expectNextCount( size ).verifyComplete();
+    }
+
     private void assertNoParallelScheduler()
     {
         Set<Thread> threadSet = Thread.getAllStackTraces().keySet();