Optimize StatementResultCursor#consumeAsync()

Previously it was implemented using `#forEachAsync()` and a no-op
consumer. This resulted in every record being added to the queue of
incoming records, then wrapped in a `CompletableFuture` and only then
thrown away. So it did a lot more work than needed.

This commit optimizes `#consumeAsync()` and `StatementResult#consume()`
that builds on top of it. Now all incoming values will not be
transformed into records and will be dropped right away. So no queueing
and `CompletableFuture`s will be involved.

Author: lutovich

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

@@ -95,9 +95,7 @@ public CompletionStage<Record> singleAsync()
     @Override
     public CompletionStage<ResultSummary> consumeAsync()
     {
-        return forEachAsync( record ->
-        {
-        } );
+        return pullAllHandler.consumeAsync();
     }
 
     @Override

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

@@ -62,6 +62,7 @@ public abstract class PullAllResponseHandler implements ResponseHandler
     private Throwable failure;
     private ResultSummary summary;
 
+    private boolean ignoreRecords;
     private CompletableFuture<Record> recordFuture;
     private CompletableFuture<Throwable> failureFuture;
 
@@ -116,9 +117,16 @@ public synchronized void onFailure( Throwable error )
     @Override
     public synchronized void onRecord( Value[] fields )
     {
-        Record record = new InternalRecord( runResponseHandler.statementKeys(), fields );
-        enqueueRecord( record );
-        completeRecordFuture( record );
+        if ( ignoreRecords )
+        {
+            completeRecordFuture( null );
+        }
+        else
+        {
+            Record record = new InternalRecord( runResponseHandler.statementKeys(), fields );
+            enqueueRecord( record );
+            completeRecordFuture( record );
+        }
     }
 
     public synchronized CompletionStage<Record> peekAsync()
@@ -131,7 +139,7 @@ public synchronized CompletionStage<Record> peekAsync()
                 return failedFuture( extractFailure() );
             }
 
-            if ( finished )
+            if ( ignoreRecords || finished )
             {
                 return completedWithNull();
             }
@@ -165,6 +173,13 @@ public synchronized CompletionStage<ResultSummary> summaryAsync()
         } );
     }
 
+    public synchronized CompletionStage<ResultSummary> consumeAsync()
+    {
+        ignoreRecords = true;
+        records.clear();
+        return summaryAsync();
+    }
+
     public synchronized <T> CompletionStage<List<T>> listAsync( Function<Record,T> mapFunction )
     {
         return failureAsync().thenApply( error ->

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

@@ -197,60 +197,6 @@ public void shouldFailWhenAskedForSingleRecordButResultContainsMore()
         }
     }
 
-    @Test
-    public void shouldConsumeAsyncWhenResultContainsMultipleRecords()
-    {
-        PullAllResponseHandler pullAllHandler = mock( PullAllResponseHandler.class );
-
-        Record record1 = new InternalRecord( asList( "key1", "key2", "key3" ), values( 1, 1, 1 ) );
-        Record record2 = new InternalRecord( asList( "key1", "key2", "key3" ), values( 2, 2, 2 ) );
-        Record record3 = new InternalRecord( asList( "key1", "key2", "key3" ), values( 3, 3, 3 ) );
-        when( pullAllHandler.nextAsync() ).thenReturn( completedFuture( record1 ) )
-                .thenReturn( completedFuture( record2 ) ).thenReturn( completedFuture( record3 ) )
-                .thenReturn( completedWithNull() );
-
-        ResultSummary summary = mock( ResultSummary.class );
-        when( pullAllHandler.summaryAsync() ).thenReturn( completedFuture( summary ) );
-
-        InternalStatementResultCursor cursor = newCursor( pullAllHandler );
-
-        assertEquals( summary, await( cursor.consumeAsync() ) );
-        verify( pullAllHandler, times( 4 ) ).nextAsync();
-    }
-
-    @Test
-    public void shouldConsumeAsyncWhenResultContainsOneRecords()
-    {
-        PullAllResponseHandler pullAllHandler = mock( PullAllResponseHandler.class );
-
-        Record record = new InternalRecord( asList( "key1", "key2" ), values( 1, 1 ) );
-        when( pullAllHandler.nextAsync() ).thenReturn( completedFuture( record ) )
-                .thenReturn( completedWithNull() );
-
-        ResultSummary summary = mock( ResultSummary.class );
-        when( pullAllHandler.summaryAsync() ).thenReturn( completedFuture( summary ) );
-
-        InternalStatementResultCursor cursor = newCursor( pullAllHandler );
-
-        assertEquals( summary, await( cursor.consumeAsync() ) );
-        verify( pullAllHandler, times( 2 ) ).nextAsync();
-    }
-
-    @Test
-    public void shouldConsumeAsyncWhenResultContainsNoRecords()
-    {
-        PullAllResponseHandler pullAllHandler = mock( PullAllResponseHandler.class );
-        when( pullAllHandler.nextAsync() ).thenReturn( completedWithNull() );
-
-        ResultSummary summary = mock( ResultSummary.class );
-        when( pullAllHandler.summaryAsync() ).thenReturn( completedFuture( summary ) );
-
-        InternalStatementResultCursor cursor = newCursor( pullAllHandler );
-
-        assertEquals( summary, await( cursor.consumeAsync() ) );
-        verify( pullAllHandler ).nextAsync();
-    }
-
     @Test
     public void shouldForEachAsyncWhenResultContainsMultipleRecords()
     {
@@ -455,6 +401,38 @@ public void shouldPropagateFailureFromListAsyncWithMapFunction()
         verify( pullAllHandler ).listAsync( mapFunction );
     }
 
+    @Test
+    public void shouldConsumeAsync()
+    {
+        PullAllResponseHandler pullAllHandler = mock( PullAllResponseHandler.class );
+        ResultSummary summary = mock( ResultSummary.class );
+        when( pullAllHandler.consumeAsync() ).thenReturn( completedFuture( summary ) );
+
+        InternalStatementResultCursor cursor = newCursor( pullAllHandler );
+
+        assertEquals( summary, await( cursor.consumeAsync() ) );
+    }
+
+    @Test
+    public void shouldPropagateFailureInConsumeAsync()
+    {
+        PullAllResponseHandler pullAllHandler = mock( PullAllResponseHandler.class );
+        RuntimeException error = new RuntimeException( "Hi" );
+        when( pullAllHandler.consumeAsync() ).thenReturn( failedFuture( error ) );
+
+        InternalStatementResultCursor cursor = newCursor( pullAllHandler );
+
+        try
+        {
+            await( cursor.consumeAsync() );
+            fail( "Exception expected" );
+        }
+        catch ( RuntimeException e )
+        {
+            assertEquals( error, e );
+        }
+    }
+
     private static InternalStatementResultCursor newCursor( PullAllResponseHandler pullAllHandler )
     {
         return new InternalStatementResultCursor( new RunResponseHandler( new CompletableFuture<>() ), pullAllHandler );