Changed session.run not discard previous run result, but buffer all into memory.

This enables nested session runs.

Author: Zhen Li

driver/src/main/java/org/neo4j/driver/StatementResult.java

@@ -143,8 +143,7 @@ public interface StatementResult extends Iterator<Record>
      *
      * If the records in the result is not fully consumed, then calling this method will exhausts the result.
      *
-     * If you want to obtain the summary without discard the records, invoke
-     * {@link StatementResult#list()} before calling this method to buffer all records into memory.
+     * If you want to access unconsumed records after summary, you shall use {@link StatementResult#list()} to buffer all records into memory before summary.
      *
      * @return a summary for the whole query result.
      */

driver/src/main/java/org/neo4j/driver/async/StatementResultCursor.java

@@ -27,6 +27,7 @@
 
 import org.neo4j.driver.Record;
 import org.neo4j.driver.Records;
+import org.neo4j.driver.StatementResult;
 import org.neo4j.driver.exceptions.NoSuchRecordException;
 import org.neo4j.driver.summary.ResultSummary;
 
@@ -74,8 +75,7 @@ public interface StatementResultCursor
      * <p>
      * If the records in the result is not fully consumed, then calling this method will exhausts the result.
      * <p>
-     * If you want to obtain the summary without discarding the records, invoke {@link #listAsync()}
-     * to buffer records into memory before calling this method.
+     * If you want to access unconsumed records after summary, you shall use {@link StatementResult#list()} to buffer all records into memory before summary.
      *
      * @return a {@link CompletionStage} completed with a summary for the whole query result. Stage can also be
      * completed exceptionally if query execution fails.

driver/src/main/java/org/neo4j/driver/internal/FailableCursor.java

@@ -22,5 +22,6 @@
 
 public interface FailableCursor
 {
+    CompletionStage<Throwable> consumeAsync();
     CompletionStage<Throwable> failureAsync();
 }

driver/src/main/java/org/neo4j/driver/internal/async/NetworkSession.java

@@ -194,7 +194,7 @@ public CompletionStage<Void> closeAsync()
                 if ( cursor != null )
                 {
                     // there exists a cursor with potentially unconsumed error, try to extract and propagate it
-                    return cursor.failureAsync();
+                    return cursor.consumeAsync();
                 }
                 // no result cursor exists so no error exists
                 return completedWithNull();

driver/src/main/java/org/neo4j/driver/internal/async/ResultCursorsHolder.java

@@ -74,6 +74,6 @@ private static CompletionStage<Throwable> retrieveFailure( CompletionStage<? ext
     {
         return cursorStage
                 .exceptionally( cursor -> null )
-                .thenCompose( cursor -> cursor == null ? completedWithNull() : cursor.failureAsync() );
+                .thenCompose( cursor -> cursor == null ? completedWithNull() : cursor.consumeAsync() );
     }
 }

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

@@ -111,11 +111,18 @@ public <T> CompletionStage<List<T>> listAsync( Function<Record,T> mapFunction )
     }
 
     @Override
-    public CompletionStage<Throwable> failureAsync()
+    public CompletionStage<Throwable> consumeAsync()
     {
         return pullAllHandler.summaryAsync().handle( ( summary, error ) -> error );
     }
 
+    @Override
+    public CompletionStage<Throwable> failureAsync()
+    {
+        return pullAllHandler.failureAsync();
+    }
+
+
     private void internalForEachAsync( Consumer<Record> action, CompletableFuture<Void> resultFuture )
     {
         CompletionStage<Record> recordFuture = nextAsync();

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

@@ -91,12 +91,19 @@ public void cancel()
     }
 
     @Override
-    public CompletionStage<Throwable> failureAsync()
+    public CompletionStage<Throwable> consumeAsync()
     {
         // calling this method will enforce discarding record stream and finish running cypher query
         return summaryAsync().thenApply( summary -> (Throwable) null ).exceptionally( error -> error );
     }
 
+    @Override
+    public CompletionStage<Throwable> failureAsync()
+    {
+        // It is safe to discard records as either the streaming has not started at all, or the streaming is fully finished.
+        return consumeAsync();
+    }
+
     @Override
     public CompletionStage<ResultSummary> summaryAsync()
     {

driver/src/main/java/org/neo4j/driver/internal/handlers/PullAllResponseHandler.java

@@ -36,5 +36,7 @@ public interface PullAllResponseHandler extends ResponseHandler
 
     <T> CompletionStage<List<T>> listAsync( Function<Record, T> mapFunction );
 
+    CompletionStage<Throwable> failureAsync();
+
     void prePopulateRecords();
 }

driver/src/main/java/org/neo4j/driver/internal/handlers/pulln/AutoPullResponseHandler.java

@@ -176,6 +176,12 @@ public synchronized <T> CompletionStage<List<T>> listAsync( Function<Record,T> m
         return pullAllAsync().thenApply( summary -> recordsAsList( mapFunction ) );
     }
 
+    @Override
+    public synchronized CompletionStage<Throwable> failureAsync()
+    {
+        return pullAllAsync().handle( ( ignore, error ) -> error );
+    }
+
     @Override
     public void prePopulateRecords()
     {

driver/src/test/java/org/neo4j/driver/integration/async/AsyncSessionIT.java

@@ -73,8 +73,8 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.neo4j.driver.Values.parameters;
 import static org.neo4j.driver.SessionConfig.builder;
+import static org.neo4j.driver.Values.parameters;
 import static org.neo4j.driver.internal.util.Futures.failedFuture;
 import static org.neo4j.driver.internal.util.Iterables.single;
 import static org.neo4j.driver.internal.util.Matchers.arithmeticError;

driver/src/test/java/org/neo4j/driver/internal/async/AsyncStatementResultCursorImplTest.java

@@ -291,7 +291,7 @@ void shouldReturnFailureWhenExists()
         PullAllResponseHandler pullAllHandler = mock( PullAllResponseHandler.class );
 
         ServiceUnavailableException error = new ServiceUnavailableException( "Hi" );
-        when( pullAllHandler.summaryAsync() ).thenReturn( failedFuture( error ) );
+        when( pullAllHandler.failureAsync() ).thenReturn( completedFuture( error ) );
 
         AsyncStatementResultCursorImpl cursor = newCursor( pullAllHandler );
 
@@ -302,7 +302,7 @@ void shouldReturnFailureWhenExists()
     void shouldReturnNullFailureWhenDoesNotExist()
     {
         PullAllResponseHandler pullAllHandler = mock( PullAllResponseHandler.class );
-        when( pullAllHandler.summaryAsync() ).thenReturn( completedWithNull() );
+        when( pullAllHandler.failureAsync() ).thenReturn( completedWithNull() );
 
         AsyncStatementResultCursorImpl cursor = newCursor( pullAllHandler );
 

driver/src/test/java/org/neo4j/driver/internal/async/ResultCursorsHolderTest.java

@@ -149,7 +149,7 @@ private static CompletionStage<AsyncStatementResultCursorImpl> cursorWithError(
     private static CompletionStage<AsyncStatementResultCursorImpl> cursorWithFailureFuture( CompletableFuture<Throwable> future )
     {
         AsyncStatementResultCursorImpl cursor = mock( AsyncStatementResultCursorImpl.class );
-        when( cursor.failureAsync() ).thenReturn( future );
+        when( cursor.consumeAsync() ).thenReturn( future );
         return completedFuture( cursor );
     }
 }

driver/src/test/java/org/neo4j/driver/internal/handlers/LegacyPullAllResponseHandlerTest.java

@@ -20,16 +20,12 @@
 
 import org.junit.jupiter.api.Test;
 
-import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
 import java.util.function.Function;
 
 import org.neo4j.driver.Record;
 import org.neo4j.driver.Statement;
-import org.neo4j.driver.exceptions.ServiceUnavailableException;
-import org.neo4j.driver.exceptions.SessionExpiredException;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.summary.ResultSummary;
 
@@ -52,121 +48,6 @@
 
 class LegacyPullAllResponseHandlerTest extends PullAllResponseHandlerTestBase<LegacyPullAllResponseHandler>
 {
-    @Test
-    void shouldReturnNoFailureWhenAlreadySucceeded()
-    {
-        LegacyPullAllResponseHandler handler = newHandler();
-        handler.onSuccess( emptyMap() );
-
-        Throwable failure = await( handler.failureAsync() );
-
-        assertNull( failure );
-    }
-
-    @Test
-    void shouldReturnNoFailureWhenSucceededAfterFailureRequested()
-    {
-        LegacyPullAllResponseHandler handler = newHandler();
-
-        CompletableFuture<Throwable> failureFuture = handler.failureAsync().toCompletableFuture();
-        assertFalse( failureFuture.isDone() );
-
-        handler.onSuccess( emptyMap() );
-
-        assertTrue( failureFuture.isDone() );
-        assertNull( await( failureFuture ) );
-    }
-
-    @Test
-    void shouldReturnFailureWhenAlreadyFailed()
-    {
-        LegacyPullAllResponseHandler handler = newHandler();
-
-        RuntimeException failure = new RuntimeException( "Ops" );
-        handler.onFailure( failure );
-
-        Throwable receivedFailure = await( handler.failureAsync() );
-        assertEquals( failure, receivedFailure );
-    }
-
-    @Test
-    void shouldReturnFailureWhenFailedAfterFailureRequested()
-    {
-        LegacyPullAllResponseHandler handler = newHandler();
-
-        CompletableFuture<Throwable> failureFuture = handler.failureAsync().toCompletableFuture();
-        assertFalse( failureFuture.isDone() );
-
-        IOException failure = new IOException( "Broken pipe" );
-        handler.onFailure( failure );
-
-        assertTrue( failureFuture.isDone() );
-        assertEquals( failure, await( failureFuture ) );
-    }
-
-    @Test
-    void shouldReturnFailureWhenRequestedMultipleTimes()
-    {
-        LegacyPullAllResponseHandler handler = newHandler();
-
-        CompletableFuture<Throwable> failureFuture1 = handler.failureAsync().toCompletableFuture();
-        CompletableFuture<Throwable> failureFuture2 = handler.failureAsync().toCompletableFuture();
-
-        assertFalse( failureFuture1.isDone() );
-        assertFalse( failureFuture2.isDone() );
-
-        RuntimeException failure = new RuntimeException( "Unable to contact database" );
-        handler.onFailure( failure );
-
-        assertTrue( failureFuture1.isDone() );
-        assertTrue( failureFuture2.isDone() );
-
-        assertEquals( failure, await( failureFuture1 ) );
-        assertEquals( failure, await( failureFuture2 ) );
-    }
-
-    @Test
-    void shouldReturnFailureOnlyOnceWhenFailedBeforeFailureRequested()
-    {
-        LegacyPullAllResponseHandler handler = newHandler();
-
-        ServiceUnavailableException failure = new ServiceUnavailableException( "Connection terminated" );
-        handler.onFailure( failure );
-
-        assertEquals( failure, await( handler.failureAsync() ) );
-        assertNull( await( handler.failureAsync() ) );
-    }
-
-    @Test
-    void shouldReturnFailureOnlyOnceWhenFailedAfterFailureRequested()
-    {
-        LegacyPullAllResponseHandler handler = newHandler();
-
-        CompletionStage<Throwable> failureFuture = handler.failureAsync();
-
-        SessionExpiredException failure = new SessionExpiredException( "Network unreachable" );
-        handler.onFailure( failure );
-        assertEquals( failure, await( failureFuture ) );
-
-        assertNull( await( handler.failureAsync() ) );
-    }
-
-    @Test
-    void shouldReturnSummaryWhenAlreadyFailedAndFailureConsumed()
-    {
-        Statement statement = new Statement( "CREATE ()" );
-        LegacyPullAllResponseHandler handler = newHandler( statement );
-
-        ServiceUnavailableException failure = new ServiceUnavailableException( "Neo4j unreachable" );
-        handler.onFailure( failure );
-
-        assertEquals( failure, await( handler.failureAsync() ) );
-
-        ResultSummary summary = await( handler.summaryAsync() );
-        assertNotNull( summary );
-        assertEquals( statement, summary.statement() );
-    }
-
     @Test
     void shouldDisableAutoReadWhenTooManyRecordsArrive()
     {