Send ACK_FAILURE instead of RESET

Author: pontusmelke

driver/src/main/java/org/neo4j/driver/internal/connector/ConcurrencyGuardingConnection.java

@@ -113,6 +113,20 @@ public void reset( StreamCollector collector )
         }
     }
 
+    @Override
+    public void ackFailure( StreamCollector collector )
+    {
+        try
+        {
+            markAsInUse();
+            delegate.ackFailure( collector );
+        }
+        finally
+        {
+            markAsAvailable();
+        }
+    }
+
     @Override
     public void sync()
     {

driver/src/main/java/org/neo4j/driver/internal/connector/socket/LoggingResponseHandler.java

@@ -24,6 +24,7 @@
 import org.neo4j.driver.v1.Logger;
 import org.neo4j.driver.v1.Value;
 
+import static org.neo4j.driver.internal.messaging.AckFailureMessage.ACK_FAILURE;
 import static org.neo4j.driver.internal.messaging.DiscardAllMessage.DISCARD_ALL;
 import static org.neo4j.driver.internal.messaging.IgnoredMessage.IGNORED;
 import static org.neo4j.driver.internal.messaging.PullAllMessage.PULL_ALL;
@@ -74,6 +75,13 @@ public void handleResetMessage()
         logger.debug( DEFAULT_DEBUG_LOGGING_FORMAT, RESET );
     }
 
+    @Override
+    public void handleAckFailureMessage()
+    {
+        super.handleAckFailureMessage();
+        logger.debug( DEFAULT_DEBUG_LOGGING_FORMAT, ACK_FAILURE );
+    }
+
     @Override
     public void handleSuccessMessage( Map<String,Value> meta )
     {

driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketConnection.java

@@ -27,7 +27,6 @@
 import org.neo4j.driver.internal.messaging.InitMessage;
 import org.neo4j.driver.internal.messaging.Message;
 import org.neo4j.driver.internal.messaging.PullAllMessage;
-import org.neo4j.driver.internal.messaging.ResetMessage;
 import org.neo4j.driver.internal.messaging.RunMessage;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.v1.Logger;
@@ -37,7 +36,9 @@
 import org.neo4j.driver.v1.exceptions.ClientException;
 import org.neo4j.driver.v1.exceptions.Neo4jException;
 
+import static org.neo4j.driver.internal.messaging.AckFailureMessage.ACK_FAILURE;
 import static org.neo4j.driver.internal.messaging.DiscardAllMessage.DISCARD_ALL;
+import static org.neo4j.driver.internal.messaging.ResetMessage.RESET;
 
 public class SocketConnection implements Connection
 {
@@ -91,7 +92,13 @@ public void pullAll( StreamCollector collector )
     @Override
     public void reset( StreamCollector collector )
     {
-        queueMessage( ResetMessage.RESET, collector );
+        queueMessage( RESET, collector );
+    }
+
+    @Override
+    public void ackFailure( StreamCollector collector )
+    {
+        queueMessage( ACK_FAILURE, collector );
     }
 
     @Override
@@ -146,7 +153,7 @@ private void assertNoServerFailure()
     {
         if ( responseHandler.serverFailureOccurred() )
         {
-            reset( StreamCollector.NO_OP );
+            ackFailure( StreamCollector.NO_OP );
             Neo4jException exception = responseHandler.serverFailure();
             responseHandler.clearError();
             throw exception;

driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketResponseHandler.java

@@ -194,6 +194,12 @@ public void handleResetMessage()
 
     }
 
+    @Override
+    public void handleAckFailureMessage()
+    {
+
+    }
+
     @Override
     public void handlePullAllMessage()
     {

driver/src/main/java/org/neo4j/driver/internal/messaging/AckFailureMessage.java

@@ -0,0 +1,56 @@
+/**
+ * Copyright (c) 2002-2016 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.neo4j.driver.internal.messaging;
+
+import java.io.IOException;
+
+/**
+ * ACK_FAILURE request message
+ *
+ * This message acts as a barrier after an error, informing the server that we've seen the error
+ * message, and that messages that follow this one are safe to execute.
+ */
+public class AckFailureMessage implements Message
+{
+    public static final AckFailureMessage ACK_FAILURE= new AckFailureMessage();
+
+    @Override
+    public void dispatch( MessageHandler handler ) throws IOException
+    {
+        handler.handleAckFailureMessage();
+    }
+
+    @Override
+    public String toString()
+    {
+        return "[ACK_FAILURE]";
+    }
+
+    @Override
+    public boolean equals( Object obj )
+    {
+        return obj != null && obj.getClass() == getClass();
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return 1;
+    }
+}

driver/src/main/java/org/neo4j/driver/internal/messaging/MessageHandler.java

@@ -36,6 +36,8 @@ public interface MessageHandler
 
     void handleResetMessage() throws IOException;
 
+    void handleAckFailureMessage() throws IOException;
+
     // Responses
     void handleSuccessMessage( Map<String,Value> meta ) throws IOException;
 
@@ -44,4 +46,5 @@ public interface MessageHandler
     void handleFailureMessage( String code, String message ) throws IOException;
 
     void handleIgnoredMessage() throws IOException;
+
 }

driver/src/main/java/org/neo4j/driver/internal/messaging/PackStreamMessageFormatV1.java

@@ -56,6 +56,7 @@
 public class PackStreamMessageFormatV1 implements MessageFormat
 {
     public final static byte MSG_INIT = 0x01;
+    public final static byte MSG_ACK_FAILURE = 0x0E;
     public final static byte MSG_RESET = 0x0F;
     public final static byte MSG_RUN = 0x10;
     public final static byte MSG_DISCARD_ALL = 0x2F;
@@ -151,6 +152,13 @@ public void handleResetMessage() throws IOException
             onMessageComplete.run();
         }
 
+        @Override
+        public void handleAckFailureMessage() throws IOException
+        {
+            packer.packStructHeader( 0, MSG_ACK_FAILURE );
+            onMessageComplete.run();
+        }
+
         @Override
         public void handleSuccessMessage( Map<String,Value> meta ) throws IOException
         {

driver/src/main/java/org/neo4j/driver/internal/pool/PooledConnection.java

@@ -136,6 +136,19 @@ public void reset( StreamCollector collector )
         }
     }
 
+    @Override
+    public void ackFailure( StreamCollector collector )
+    {
+        try
+        {
+            delegate.ackFailure( collector );
+        }
+        catch ( RuntimeException e )
+        {
+            onDelegateException( e );
+        }
+    }
+
     @Override
     public void sync()
     {

driver/src/main/java/org/neo4j/driver/internal/spi/Connection.java

@@ -60,6 +60,12 @@ public interface Connection extends AutoCloseable
      */
     void reset( StreamCollector collector );
 
+    /**
+     * Queue a reset action, output will be handed to the collector once the pull starts. This will
+     * close the stream once its completed, allowing another {@link #run(String, java.util.Map, StreamCollector) run}
+     */
+    void ackFailure( StreamCollector collector );
+
     /**
      * Ensure all outstanding actions are carried out on the server.
      */

driver/src/test/java/org/neo4j/driver/v1/integration/ResultStreamIT.java

@@ -23,6 +23,7 @@
 
 import org.neo4j.driver.v1.Record;
 import org.neo4j.driver.v1.StatementResult;
+import org.neo4j.driver.v1.types.Node;
 import org.neo4j.driver.v1.util.TestNeo4jSession;
 
 import static org.junit.Assert.assertEquals;
@@ -61,11 +62,26 @@ public void shouldHaveFieldNamesInResult()
         assertEquals( "[n]", res.keys().toString() );
     }
 
+    @Test
+    public void should()
+    {
+        // When
+        StatementResult res = session.run( "CREATE (n:TestNode {name:'test', prop: [42]}) RETURN n" );
+
+        Node n = res.next().get( "n" ).asNode();
+        System.out.println( n );
+        // Then
+        assertEquals( "[n]", res.keys().toString() );
+        assertNotNull( res.single() );
+        assertEquals( "[n]", res.keys().toString() );
+    }
+
     @Test
     public void shouldGiveHelpfulFailureMessageWhenAccessNonExistingField() throws Throwable
     {
         // Given
-        StatementResult rs = session.run( "CREATE (n:Person {name:{name}}) RETURN n", parameters( "name", "Tom Hanks" ) );
+        StatementResult rs =
+                session.run( "CREATE (n:Person {name:{name}}) RETURN n", parameters( "name", "Tom Hanks" ) );
 
         // When
         Record single = rs.single();
@@ -78,7 +94,8 @@ public void shouldGiveHelpfulFailureMessageWhenAccessNonExistingField() throws T
     public void shouldGiveHelpfulFailureMessageWhenAccessNonExistingPropertyOnNode() throws Throwable
     {
         // Given
-        StatementResult rs = session.run( "CREATE (n:Person {name:{name}}) RETURN n", parameters( "name", "Tom Hanks" ) );
+        StatementResult rs =
+                session.run( "CREATE (n:Person {name:{name}}) RETURN n", parameters( "name", "Tom Hanks" ) );
 
         // When
         Record record = rs.single();
@@ -96,4 +113,26 @@ public void shouldNotReturnNullKeysOnEmptyResult()
         // THEN
         assertNotNull( rs.keys() );
     }
+
+    @Test
+    public void shouldBeAbleToReuseSessionAfterFailure() throws Throwable
+    {
+        // Given
+        StatementResult res1 = session.run( "INVALID" );
+        try
+        {
+            res1.consume();
+        }
+        catch ( Exception e )
+        {
+            //ignore
+        }
+
+        // When
+        StatementResult res2 = session.run( "RETURN 1" );
+
+        // Then
+        assertTrue( res2.hasNext() );
+        assertEquals( res2.next().get("1").asLong(), 1L );
+    }
 }