Made PULL_ALL and DISCARD_ALL response return bookmark for auto t…

…x only

On the other hand, bookmark will only return in response of `COMMIT` for explicit tx

community/bolt/src/main/java/org/neo4j/bolt/runtime/StatementProcessor.java

@@ -38,7 +38,7 @@ public interface StatementProcessor
 
     StatementMetadata run( String statement, MapValue params, Bookmark bookmark, Duration txTimeout, Map<String,Object> txMetaData ) throws KernelException;
 
-    void streamResult( ThrowingConsumer<BoltResult,Exception> resultConsumer ) throws Exception;
+    Bookmark streamResult( ThrowingConsumer<BoltResult,Exception> resultConsumer ) throws Exception;
 
     Bookmark commitTransaction() throws KernelException;
 
@@ -84,7 +84,7 @@ public StatementMetadata run( String statement, MapValue params, Bookmark bookma
         }
 
         @Override
-        public void streamResult( ThrowingConsumer<BoltResult,Exception> resultConsumer ) throws Exception
+        public Bookmark streamResult( ThrowingConsumer<BoltResult,Exception> resultConsumer ) throws Exception
         {
             throw new UnsupportedOperationException( "Unable to stream results" );
         }

community/bolt/src/main/java/org/neo4j/bolt/v1/runtime/TransactionStateMachine.java

@@ -117,14 +117,14 @@ public StatementMetadata run( String statement, MapValue params, Bookmark bookma
     }
 
     @Override
-    public void streamResult( ThrowingConsumer<BoltResult, Exception> resultConsumer ) throws Exception
+    public Bookmark streamResult( ThrowingConsumer<BoltResult, Exception> resultConsumer ) throws Exception
     {
         before();
         try
         {
             ensureNoPendingTerminationNotice();
 
-            state.streamResult( ctx, resultConsumer );
+            return state.streamResult( ctx, spi, resultConsumer );
         }
         finally
         {
@@ -323,14 +323,16 @@ void execute( MutableTransactionState ctx, TransactionStateMachineSPI spi, Strin
                     }
 
                     @Override
-                    void streamResult( MutableTransactionState ctx, ThrowingConsumer<BoltResult,Exception> resultConsumer ) throws Exception
+                    Bookmark streamResult( MutableTransactionState ctx, TransactionStateMachineSPI spi, ThrowingConsumer<BoltResult,Exception> resultConsumer )
+                            throws Exception
                     {
                         assert ctx.currentResult != null;
 
                         boolean success = false;
                         try
                         {
                             success = consumeResult( ctx, resultConsumer );
+                            return newestBookmark( spi );
                         }
                         finally
                         {
@@ -362,7 +364,7 @@ State beginTransaction( MutableTransactionState ctx, TransactionStateMachineSPI
 
                     @Override
                     State run( MutableTransactionState ctx, TransactionStateMachineSPI spi, String statement, MapValue params, Bookmark bookmark,
-                            Duration txTimeout, Map<String,Object> txMetadata )
+                            Duration ignored1, Map<String,Object> ignored2 )
                             throws KernelException
                     {
                         if ( statement.isEmpty() )
@@ -389,11 +391,12 @@ State run( MutableTransactionState ctx, TransactionStateMachineSPI spi, String s
                     }
 
                     @Override
-                    void streamResult( MutableTransactionState ctx,
-                            ThrowingConsumer<BoltResult,Exception> resultConsumer ) throws Exception
+                    Bookmark streamResult( MutableTransactionState ctx, TransactionStateMachineSPI spi, ThrowingConsumer<BoltResult,Exception> resultConsumer )
+                            throws Exception
                     {
                         assert ctx.currentResult != null;
                         consumeResult( ctx, resultConsumer );
+                        return null; // Explict tx shall not get a bookmark in PULL_ALL or DISCARD_ALL
                     }
 
                     @Override
@@ -421,7 +424,8 @@ abstract State run( MutableTransactionState ctx, TransactionStateMachineSPI spi,
                 Duration txTimeout, Map<String,Object> txMetadata )
                 throws KernelException;
 
-        abstract void streamResult( MutableTransactionState ctx, ThrowingConsumer<BoltResult,Exception> resultConsumer ) throws Exception;
+        abstract Bookmark streamResult( MutableTransactionState ctx, TransactionStateMachineSPI spi, ThrowingConsumer<BoltResult,Exception> resultConsumer )
+                throws Exception;
 
         abstract State commitTransaction( MutableTransactionState ctx, TransactionStateMachineSPI spi ) throws KernelException;
 

community/bolt/src/main/java/org/neo4j/bolt/v3/BoltStateMachineV3.java

@@ -25,12 +25,13 @@
 import org.neo4j.bolt.runtime.BoltStateMachineSPI;
 import org.neo4j.bolt.v1.runtime.BoltStateMachineV1;
 import org.neo4j.bolt.v1.runtime.InterruptedState;
-import org.neo4j.bolt.v3.runtime.FailedState;
 import org.neo4j.bolt.v3.runtime.DefunctState;
 import org.neo4j.bolt.v3.runtime.ExtraMetaDataConnectedState;
+import org.neo4j.bolt.v3.runtime.FailedState;
 import org.neo4j.bolt.v3.runtime.ReadyState;
 import org.neo4j.bolt.v3.runtime.StreamingState;
 import org.neo4j.bolt.v3.runtime.TransactionReadyState;
+import org.neo4j.bolt.v3.runtime.TransactionStreamingState;
 
 public class BoltStateMachineV3 extends BoltStateMachineV1
 {
@@ -49,7 +50,7 @@ protected States buildStates()
         InterruptedState interrupted = new InterruptedState();
         DefunctState defunct = new DefunctState();
         TransactionReadyState txReady = new TransactionReadyState();
-        StreamingState txStreaming = new StreamingState();
+        TransactionStreamingState txStreaming = new TransactionStreamingState();
 
         connected.setReadyState( ready );
         connected.setFailedState( defunct );

community/bolt/src/main/java/org/neo4j/bolt/v3/runtime/AbstractStreamingState.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright (c) 2002-2018 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.bolt.v3.runtime;
+
+import org.neo4j.bolt.messaging.RequestMessage;
+import org.neo4j.bolt.runtime.BoltConnectionFatality;
+import org.neo4j.bolt.runtime.BoltStateMachineState;
+import org.neo4j.bolt.runtime.StateMachineContext;
+import org.neo4j.bolt.v1.messaging.request.DiscardAllMessage;
+import org.neo4j.bolt.v1.messaging.request.InterruptSignal;
+import org.neo4j.bolt.v1.messaging.request.PullAllMessage;
+
+import static org.neo4j.util.Preconditions.checkState;
+
+/**
+ * When STREAMING, a result is available as a stream of records.
+ * These must be PULLed or DISCARDed before any further statements
+ * can be executed.
+ */
+public abstract class AbstractStreamingState extends FailSafeBoltStateMachineState
+{
+    protected BoltStateMachineState readyState;
+    private BoltStateMachineState interruptedState;
+
+    @Override
+    public BoltStateMachineState process( RequestMessage message, StateMachineContext context ) throws BoltConnectionFatality
+    {
+        assertInitialized();
+        if ( message instanceof PullAllMessage )
+        {
+            return processPullAllMessage( context );
+        }
+        if ( message instanceof DiscardAllMessage )
+        {
+            return processDiscardAllMessage( context );
+        }
+        if ( message instanceof InterruptSignal )
+        {
+            return interruptedState;
+        }
+        return null;
+    }
+
+    public void setReadyState( BoltStateMachineState readyState )
+    {
+        this.readyState = readyState;
+    }
+
+    public void setInterruptedState( BoltStateMachineState interruptedState )
+    {
+        this.interruptedState = interruptedState;
+    }
+
+    private BoltStateMachineState processPullAllMessage( StateMachineContext context ) throws BoltConnectionFatality
+    {
+        return processMessage( context, () -> processStreamResultMessage( true, context ) );
+    }
+
+    private BoltStateMachineState processDiscardAllMessage( StateMachineContext context ) throws BoltConnectionFatality
+    {
+        return processMessage( context, () -> processStreamResultMessage( false, context ) );
+    }
+
+    abstract BoltStateMachineState processStreamResultMessage( boolean pull, StateMachineContext context ) throws Throwable;
+
+    private void assertInitialized()
+    {
+        checkState( readyState != null, "Ready state not set" );
+        checkState( interruptedState != null, "Interrupted state not set" );
+        checkState( failedState != null, "Failed state not set" );
+    }
+}

community/bolt/src/main/java/org/neo4j/bolt/v3/runtime/StreamingState.java

@@ -19,82 +19,27 @@
  */
 package org.neo4j.bolt.v3.runtime;
 
-import org.neo4j.bolt.messaging.RequestMessage;
-import org.neo4j.bolt.runtime.BoltConnectionFatality;
 import org.neo4j.bolt.runtime.BoltStateMachineState;
 import org.neo4j.bolt.runtime.StateMachineContext;
-import org.neo4j.bolt.v1.messaging.request.DiscardAllMessage;
-import org.neo4j.bolt.v1.messaging.request.InterruptSignal;
-import org.neo4j.bolt.v1.messaging.request.PullAllMessage;
-
-import static org.neo4j.util.Preconditions.checkState;
+import org.neo4j.bolt.v1.runtime.bookmarking.Bookmark;
 
 /**
- * When STREAMING, a result is available as a stream of records.
- * These must be PULLed or DISCARDed before any further statements
- * can be executed.
+ * When STREAMING, additionally attach bookmark to PULL_ALL, DISCARD_ALL result
  */
-public class StreamingState extends FailSafeBoltStateMachineState
+public class StreamingState extends AbstractStreamingState
 {
-    private BoltStateMachineState readyState;
-    private BoltStateMachineState interruptedState;
-
-    @Override
-    public BoltStateMachineState process( RequestMessage message, StateMachineContext context ) throws BoltConnectionFatality
-    {
-        assertInitialized();
-        if ( message instanceof PullAllMessage )
-        {
-            return processPullAllMessage( context );
-        }
-        if ( message instanceof DiscardAllMessage )
-        {
-            return processDiscardAllMessage( context );
-        }
-        if ( message instanceof InterruptSignal )
-        {
-            return interruptedState;
-        }
-        return null;
-    }
-
     @Override
     public String name()
     {
         return "STREAMING";
     }
 
-    public void setReadyState( BoltStateMachineState readyState )
-    {
-        this.readyState = readyState;
-    }
-
-    public void setInterruptedState( BoltStateMachineState interruptedState )
-    {
-        this.interruptedState = interruptedState;
-    }
-
-    private BoltStateMachineState processPullAllMessage( StateMachineContext context ) throws BoltConnectionFatality
-    {
-        return processMessage( context, () -> processStreamResultMessage( true, context ) );
-    }
-
-    private BoltStateMachineState processDiscardAllMessage( StateMachineContext context ) throws BoltConnectionFatality
-    {
-        return processMessage( context, () -> processStreamResultMessage( false, context ) );
-    }
-
-    private BoltStateMachineState processStreamResultMessage( boolean pull, StateMachineContext context ) throws Throwable
+    @Override
+    protected BoltStateMachineState processStreamResultMessage( boolean pull, StateMachineContext context ) throws Throwable
     {
-        context.connectionState().getStatementProcessor().streamResult(
+        Bookmark bookmark = context.connectionState().getStatementProcessor().streamResult(
                 recordStream -> context.connectionState().getResponseHandler().onRecords( recordStream, pull ) );
+        bookmark.attachTo( context.connectionState() );
         return readyState;
     }
-
-    private void assertInitialized()
-    {
-        checkState( readyState != null, "Ready state not set" );
-        checkState( interruptedState != null, "Interrupted state not set" );
-        checkState( failedState != null, "Failed state not set" );
-    }
 }

community/bolt/src/main/java/org/neo4j/bolt/v3/runtime/TransactionReadyState.java

@@ -101,11 +101,16 @@ private BoltStateMachineState processRunMessage( RunMessage message, StateMachin
     }
 
     private BoltStateMachineState processCommitMessage( StateMachineContext context ) throws Exception
+    {
+        appendBookmarkInResponse( context );
+        return readyState;
+    }
+
+    static void appendBookmarkInResponse( StateMachineContext context ) throws KernelException
     {
         StatementProcessor statementProcessor = context.connectionState().getStatementProcessor();
         Bookmark bookmark = statementProcessor.commitTransaction();
         bookmark.attachTo( context.connectionState() );
-        return readyState;
     }
 
     private BoltStateMachineState processRollbackMessage( StateMachineContext context ) throws Exception

community/bolt/src/main/java/org/neo4j/bolt/v3/runtime/TransactionStreamingState.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2002-2018 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.bolt.v3.runtime;
+
+import org.neo4j.bolt.runtime.BoltStateMachineState;
+import org.neo4j.bolt.runtime.StateMachineContext;
+
+public class TransactionStreamingState extends AbstractStreamingState
+{
+    @Override
+    public String name()
+    {
+        return "TX_STREAMING";
+    }
+
+    protected BoltStateMachineState processStreamResultMessage( boolean pull, StateMachineContext context ) throws Throwable
+    {
+        context.connectionState().getStatementProcessor().streamResult(
+                recordStream -> context.connectionState().getResponseHandler().onRecords( recordStream, pull ) );
+        return readyState;
+    }
+}

community/bolt/src/test/java/org/neo4j/bolt/v3/messaging/request/RunMessageTest.java

@@ -42,7 +42,7 @@ class RunMessageTest
     void shouldParseEmptyTransactionMetadataCorrectly() throws Throwable
     {
         // When
-        RunMessage message = new RunMessage("RETURN 1" );
+        RunMessage message = new RunMessage( "RETURN 1" );
 
         // Then
         assertNull( message.transactionMetadata() );