Made Statement#summary to cancel streaming if streaming is still ongoing.

As a result `StatementResult#consume` is removed as it is the same as `StatementResult#summary`
Added tests back.

Author: Zhen Li

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

@@ -20,11 +20,11 @@
 
 import java.util.Iterator;
 import java.util.List;
+import java.util.function.Function;
 import java.util.stream.Stream;
 
 import org.neo4j.driver.exceptions.NoSuchRecordException;
 import org.neo4j.driver.summary.ResultSummary;
-import java.util.function.Function;
 import org.neo4j.driver.util.Resource;
 
 
@@ -138,29 +138,13 @@ public interface StatementResult extends Iterator<Record>
      */
     <T> List<T> list( Function<Record, T> mapFunction );
 
-    /**
-     * Consume the entire result, yielding a summary of it.
-     *
-     * Calling this method exhausts the result.
-     *
-     * <pre class="doctest:ResultDocIT#summarizeUsage">
-     * {@code
-     * ResultSummary summary = session.run( "PROFILE MATCH (n:User {id: 12345}) RETURN n" ).consume();
-     * }
-     * </pre>
-     *
-     * @return a summary for the whole query result
-     */
-    ResultSummary consume();
-
     /**
      * Return the result summary.
      *
-     * If the records in the result is not fully consumed, then calling this method will force to pull all remaining
-     * records into buffer to yield the summary.
+     * 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 but discard the records, use
-     * {@link StatementResult#consume()} instead.
+     * 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.
      *
      * @return a summary for the whole query result.
      */

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

@@ -72,10 +72,10 @@ public interface StatementResultCursor
     /**
      * Asynchronously retrieve the result summary.
      * <p>
-     * If the records in the result is not fully consumed, then calling this method will force to pull all remaining
-     * records into buffer to yield the summary.
+     * 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 but discard the records, use {@link #consumeAsync()} instead.
+     * If you want to obtain the summary without discarding the records, invoke {@link #listAsync()}
+     * to buffer records into memory before calling this method.
      *
      * @return a {@link CompletionStage} completed with a summary for the whole query result. Stage can also be
      * completed exceptionally if query execution fails.
@@ -110,14 +110,6 @@ public interface StatementResultCursor
      */
     CompletionStage<Record> singleAsync();
 
-    /**
-     * Asynchronously consume the entire result, yielding a summary of it. Calling this method exhausts the result.
-     *
-     * @return a {@link CompletionStage} completed with a summary for the whole query result. Stage can also be
-     * completed exceptionally if query execution fails.
-     */
-    CompletionStage<ResultSummary> consumeAsync();
-
     /**
      * Asynchronously apply the given {@link Consumer action} to every record in the result, yielding a summary of it.
      *

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

@@ -22,18 +22,18 @@
 import java.util.Spliterator;
 import java.util.Spliterators;
 import java.util.concurrent.CompletionStage;
+import java.util.function.Function;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
-import org.neo4j.driver.internal.spi.Connection;
-import org.neo4j.driver.internal.util.Futures;
 import org.neo4j.driver.Record;
 import org.neo4j.driver.StatementResult;
 import org.neo4j.driver.async.StatementResultCursor;
 import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.exceptions.NoSuchRecordException;
+import org.neo4j.driver.internal.spi.Connection;
+import org.neo4j.driver.internal.util.Futures;
 import org.neo4j.driver.summary.ResultSummary;
-import java.util.function.Function;
 
 public class InternalStatementResult implements StatementResult
 {
@@ -111,12 +111,6 @@ public <T> List<T> list( Function<Record, T> mapFunction )
         return blockingGet( cursor.listAsync( mapFunction ) );
     }
 
-    @Override
-    public ResultSummary consume()
-    {
-        return blockingGet( cursor.consumeAsync() );
-    }
-
     @Override
     public ResultSummary summary()
     {

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

@@ -29,7 +29,7 @@
 import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.internal.BookmarkHolder;
 import org.neo4j.driver.internal.InternalBookmark;
-import org.neo4j.driver.internal.cursor.InternalStatementResultCursor;
+import org.neo4j.driver.internal.cursor.AsyncStatementResultCursor;
 import org.neo4j.driver.internal.cursor.RxStatementResultCursor;
 import org.neo4j.driver.internal.messaging.BoltProtocol;
 import org.neo4j.driver.internal.spi.Connection;
@@ -139,7 +139,7 @@ else if ( state == State.ROLLED_BACK )
     public CompletionStage<StatementResultCursor> runAsync( Statement statement, boolean waitForRunResponse )
     {
         ensureCanRunQueries();
-        CompletionStage<InternalStatementResultCursor> cursorStage =
+        CompletionStage<AsyncStatementResultCursor> cursorStage =
                 protocol.runInExplicitTransaction( connection, statement, this, waitForRunResponse ).asyncResult();
         resultCursors.add( cursorStage );
         return cursorStage.thenApply( cursor -> cursor );

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

@@ -34,7 +34,7 @@
 import org.neo4j.driver.internal.DatabaseName;
 import org.neo4j.driver.internal.FailableCursor;
 import org.neo4j.driver.internal.InternalBookmark;
-import org.neo4j.driver.internal.cursor.InternalStatementResultCursor;
+import org.neo4j.driver.internal.cursor.AsyncStatementResultCursor;
 import org.neo4j.driver.internal.cursor.RxStatementResultCursor;
 import org.neo4j.driver.internal.cursor.StatementResultCursorFactory;
 import org.neo4j.driver.internal.logging.PrefixedLogger;
@@ -76,7 +76,7 @@ public NetworkSession( ConnectionProvider connectionProvider, RetryLogic retryLo
 
     public CompletionStage<StatementResultCursor> runAsync( Statement statement, TransactionConfig config, boolean waitForRunResponse )
     {
-        CompletionStage<InternalStatementResultCursor> newResultCursorStage =
+        CompletionStage<AsyncStatementResultCursor> newResultCursorStage =
                 buildResultCursorFactory( statement, config, waitForRunResponse ).thenCompose( StatementResultCursorFactory::asyncResult );
 
         resultCursorStage = newResultCursorStage.exceptionally( error -> null );

driver/src/main/java/org/neo4j/driver/internal/cursor/InternalStatementResultCursor.java renamed to driver/src/main/java/org/neo4j/driver/internal/cursor/AsyncStatementResultCursor.java

@@ -21,6 +21,6 @@
 import org.neo4j.driver.internal.FailableCursor;
 import org.neo4j.driver.async.StatementResultCursor;
 
-public interface InternalStatementResultCursor extends StatementResultCursor, FailableCursor
+public interface AsyncStatementResultCursor extends StatementResultCursor, FailableCursor
 {
 }

driver/src/main/java/org/neo4j/driver/internal/async/AsyncStatementResultCursor.java renamed to driver/src/main/java/org/neo4j/driver/internal/cursor/AsyncStatementResultCursorImpl.java

@@ -16,28 +16,27 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.neo4j.driver.internal.async;
+package org.neo4j.driver.internal.cursor;
 
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import org.neo4j.driver.Record;
+import org.neo4j.driver.exceptions.NoSuchRecordException;
 import org.neo4j.driver.internal.handlers.PullAllResponseHandler;
 import org.neo4j.driver.internal.handlers.RunResponseHandler;
 import org.neo4j.driver.internal.util.Futures;
-import org.neo4j.driver.internal.cursor.InternalStatementResultCursor;
-import org.neo4j.driver.Record;
-import org.neo4j.driver.exceptions.NoSuchRecordException;
 import org.neo4j.driver.summary.ResultSummary;
 
-public class AsyncStatementResultCursor implements InternalStatementResultCursor
+public class AsyncStatementResultCursorImpl implements AsyncStatementResultCursor
 {
     private final RunResponseHandler runHandler;
     private final PullAllResponseHandler pullAllHandler;
 
-    public AsyncStatementResultCursor( RunResponseHandler runHandler, PullAllResponseHandler pullAllHandler )
+    public AsyncStatementResultCursorImpl( RunResponseHandler runHandler, PullAllResponseHandler pullAllHandler )
     {
         this.runHandler = runHandler;
         this.pullAllHandler = pullAllHandler;
@@ -91,12 +90,6 @@ public CompletionStage<Record> singleAsync()
         } );
     }
 
-    @Override
-    public CompletionStage<ResultSummary> consumeAsync()
-    {
-        return pullAllHandler.summaryAsync();
-    }
-
     @Override
     public CompletionStage<ResultSummary> forEachAsync( Consumer<Record> action )
     {

driver/src/main/java/org/neo4j/driver/internal/cursor/AsyncResultCursorOnlyFactory.java renamed to driver/src/main/java/org/neo4j/driver/internal/cursor/AsyncStatementResultCursorOnlyFactory.java

@@ -20,30 +20,28 @@
 
 import java.util.concurrent.CompletionStage;
 
-import org.neo4j.driver.internal.async.AsyncStatementResultCursor;
+import org.neo4j.driver.exceptions.ClientException;
 import org.neo4j.driver.internal.handlers.PullAllResponseHandler;
 import org.neo4j.driver.internal.handlers.RunResponseHandler;
 import org.neo4j.driver.internal.messaging.Message;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.util.Futures;
-import org.neo4j.driver.exceptions.ClientException;
 
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.CompletableFuture.completedFuture;
-import static org.neo4j.driver.internal.messaging.request.PullAllMessage.PULL_ALL;
 
 /**
  * Used by Bolt V1, V2, V3
  */
-public class AsyncResultCursorOnlyFactory implements StatementResultCursorFactory
+public class AsyncStatementResultCursorOnlyFactory implements StatementResultCursorFactory
 {
     protected final Connection connection;
     protected final Message runMessage;
     protected final RunResponseHandler runHandler;
     protected final PullAllResponseHandler pullAllHandler;
     private final boolean waitForRunResponse;
 
-    public AsyncResultCursorOnlyFactory( Connection connection, Message runMessage, RunResponseHandler runHandler,
+    public AsyncStatementResultCursorOnlyFactory( Connection connection, Message runMessage, RunResponseHandler runHandler,
             PullAllResponseHandler pullHandler, boolean waitForRunResponse )
     {
         requireNonNull( connection );
@@ -59,19 +57,20 @@ public AsyncResultCursorOnlyFactory( Connection connection, Message runMessage,
         this.waitForRunResponse = waitForRunResponse;
     }
 
-    public CompletionStage<InternalStatementResultCursor> asyncResult()
+    public CompletionStage<AsyncStatementResultCursor> asyncResult()
     {
         // only write and flush messages when async result is wanted.
-        connection.writeAndFlush( runMessage, runHandler, PULL_ALL, pullAllHandler );
+        connection.writeAndFlush( runMessage, runHandler );
+        pullAllHandler.prePopulateRecords();
 
         if ( waitForRunResponse )
         {
             // wait for response of RUN before proceeding
-            return runHandler.runFuture().thenApply( ignore -> new AsyncStatementResultCursor( runHandler, pullAllHandler ) );
+            return runHandler.runFuture().thenApply( ignore -> new AsyncStatementResultCursorImpl( runHandler, pullAllHandler ) );
         }
         else
         {
-            return completedFuture( new AsyncStatementResultCursor( runHandler, pullAllHandler ) );
+            return completedFuture( new AsyncStatementResultCursorImpl( runHandler, pullAllHandler ) );
         }
     }
 

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

@@ -21,129 +21,20 @@
 import org.reactivestreams.Subscription;
 
 import java.util.List;
-import java.util.Objects;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiConsumer;
 
-import org.neo4j.driver.internal.FailableCursor;
-import org.neo4j.driver.internal.handlers.RunResponseHandler;
-import org.neo4j.driver.internal.handlers.pulln.PullResponseHandler;
 import org.neo4j.driver.Record;
+import org.neo4j.driver.internal.FailableCursor;
 import org.neo4j.driver.summary.ResultSummary;
 
-import static org.neo4j.driver.internal.handlers.pulln.BasicPullResponseHandler.DISCARD_RECORD_CONSUMER;
-
-public class RxStatementResultCursor implements Subscription, FailableCursor
+public interface RxStatementResultCursor extends Subscription, FailableCursor
 {
-    private final RunResponseHandler runHandler;
-    private final PullResponseHandler pullHandler;
-    private final Throwable runResponseError;
-    private final CompletableFuture<ResultSummary> summaryFuture = new CompletableFuture<>();
-    private BiConsumer<Record,Throwable> recordConsumer;
-
-    public RxStatementResultCursor( RunResponseHandler runHandler, PullResponseHandler pullHandler )
-    {
-        this( null, runHandler, pullHandler );
-    }
-
-    public RxStatementResultCursor( Throwable runError, RunResponseHandler runHandler, PullResponseHandler pullHandler )
-    {
-        Objects.requireNonNull( runHandler );
-        Objects.requireNonNull( pullHandler );
-        assertRunResponseArrived( runHandler );
-
-        this.runResponseError = runError;
-        this.runHandler = runHandler;
-        this.pullHandler = pullHandler;
-        installSummaryConsumer();
-    }
-
-    public List<String> keys()
-    {
-        return runHandler.statementKeys();
-    }
-
-    public void installRecordConsumer( BiConsumer<Record,Throwable> recordConsumer )
-    {
-        if ( isRecordConsumerInstalled() )
-        {
-            return;
-        }
-        this.recordConsumer = recordConsumer;
-        pullHandler.installRecordConsumer( this.recordConsumer );
-        assertRunCompletedSuccessfully();
-    }
-
-    private boolean isRecordConsumerInstalled()
-    {
-        return this.recordConsumer != null;
-    }
-
-    public void request( long n )
-    {
-        pullHandler.request( n );
-    }
-
-    @Override
-    public void cancel()
-    {
-        pullHandler.cancel();
-    }
-
-    @Override
-    public CompletionStage<Throwable> failureAsync()
-    {
-        // calling this method will enforce discarding record stream and finish running cypher query
-        return summaryAsync().thenApply( summary -> (Throwable) null ).exceptionally( error -> error );
-    }
-
-    public CompletionStage<ResultSummary> summaryAsync()
-    {
-        if ( !isDone() ) // the summary is called before record streaming
-        {
-            installRecordConsumer( DISCARD_RECORD_CONSUMER );
-            cancel();
-        }
-
-        return this.summaryFuture;
-    }
-
-    public boolean isDone()
-    {
-        return summaryFuture.isDone();
-    }
+    List<String> keys();
 
-    private void assertRunCompletedSuccessfully()
-    {
-        if ( runResponseError != null )
-        {
-            pullHandler.onFailure( runResponseError );
-        }
-    }
+    void installRecordConsumer( BiConsumer<Record,Throwable> recordConsumer );
 
-    private void installSummaryConsumer()
-    {
-        pullHandler.installSummaryConsumer( ( summary, error ) -> {
-            if ( error != null && recordConsumer == DISCARD_RECORD_CONSUMER )
-            {
-                // We will only report the error to summary if there is no user record consumer installed
-                // When a user record consumer is installed, the error will be reported to record consumer instead.
-                summaryFuture.completeExceptionally( error );
-            }
-            else if ( summary != null )
-            {
-                summaryFuture.complete( summary );
-            }
-            //else (null, null) to indicate a has_more success
-        } );
-    }
+    CompletionStage<ResultSummary> summaryAsync();
 
-    private void assertRunResponseArrived( RunResponseHandler runHandler )
-    {
-        if ( !runHandler.runFuture().isDone() )
-        {
-            throw new IllegalStateException( "Should wait for response of RUN before allowing PULL." );
-        }
-    }
+    boolean isDone();
 }