Merge branch '1.0' into 1.1

Author: Zhen

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

@@ -148,9 +148,14 @@ public void run()
                 //must check if transaction has been closed
                 if (currentTransaction != null)
                 {
-                    currentTransaction.markAsRolledBack();
-                    currentTransaction = null;
-                    connection.onError( null );
+                    if( connection.hasUnrecoverableErrors() )
+                    {
+                        currentTransaction.markToClose();
+                    }
+                    else
+                    {
+                        currentTransaction.failure();
+                    }
                 }
             }
         });
@@ -165,12 +170,14 @@ public TypeSystem typeSystem()
 
     private void ensureConnectionIsValidBeforeRunningSession()
     {
+        ensureNoUnrecoverableError();
         ensureNoOpenTransactionBeforeRunningSession();
         ensureConnectionIsOpen();
     }
 
     private void ensureConnectionIsValidBeforeOpeningTransaction()
     {
+        ensureNoUnrecoverableError();
         ensureNoOpenTransactionBeforeOpeningTransaction();
         ensureConnectionIsOpen();
     }
@@ -187,6 +194,16 @@ protected void finalize() throws Throwable
         super.finalize();
     }
 
+    private void ensureNoUnrecoverableError()
+    {
+        if( connection.hasUnrecoverableErrors() )
+        {
+            throw new ClientException( "Cannot run more statements in the current session as an unrecoverable error " +
+                                       "has happened. Please close the current session and re-run your statement in a" +
+                                       " new session." );
+        }
+    }
+
     private void ensureNoOpenTransactionBeforeRunningSession()
     {
         if ( currentTransaction != null )

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

@@ -66,32 +66,14 @@ public InternalStatementResult( Connection connection, Statement statement )
 
     private StreamCollector newRunResponseCollector()
     {
-        return new StreamCollector()
+        return new StreamCollector.NoOperationStreamCollector()
         {
             @Override
             public void keys( String[] names )
             {
                 keys = Arrays.asList( names );
             }
 
-            @Override
-            public void record( Value[] fields ) {}
-
-            @Override
-            public void statementType( StatementType type ) {}
-
-            @Override
-            public void statementStatistics( SummaryCounters statistics ) {}
-
-            @Override
-            public void plan( Plan plan ) {}
-
-            @Override
-            public void profile( ProfiledPlan plan ) {}
-
-            @Override
-            public void notifications( List<Notification> notifications ) {}
-
             @Override
             public void done()
             {
@@ -106,11 +88,8 @@ public void done()
     private StreamCollector newPullAllResponseCollector( Statement statement )
     {
         final SummaryBuilder summaryBuilder = new SummaryBuilder( statement );
-        return new StreamCollector()
+        return new StreamCollector.NoOperationStreamCollector()
         {
-            @Override
-            public void keys( String[] names ) {}
-
             @Override
             public void record( Value[] fields )
             {

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

@@ -112,8 +112,6 @@ public void close()
                 }
                 else if ( state == State.MARKED_FAILED || state == State.ACTIVE )
                 {
-                    // If alwaysValid of the things we've put in the queue have been sent off, there is no need to
-                    // do this, we could just clear the queue. Future optimization.
                     conn.run( "ROLLBACK", Collections.<String, Value>emptyMap(), StreamCollector.NO_OP );
                     conn.discardAll();
                     conn.sync();
@@ -171,6 +169,8 @@ public StatementResult run( Statement statement )
         }
         catch ( Neo4jException e )
         {
+            // Failed to send messages to the server probably due to IOException in the socket.
+            // So we should stop sending more messages in this transaction
             state = State.FAILED;
             throw e;
         }
@@ -200,12 +200,8 @@ public TypeSystem typeSystem()
         return InternalTypeSystem.TYPE_SYSTEM;
     }
 
-    // TODO: This is wrong. This is only needed because we changed the SSM
-    // to move to IDLE on any exception (so the normal `ROLLBACK` statement won't work).
-    // We should change the SSM to move to some special ROLLBACK_ONLY state instead and
-    // remove this code path
-    public void markAsRolledBack()
+    public void markToClose()
     {
-        state = State.ROLLED_BACK;
+        state = State.FAILED;
     }
 }

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
         {
@@ -319,7 +327,7 @@ private void packValue( Value value ) throws IOException
                     break;
 
                 default:
-                    throw new UnsupportedOperationException( "Unknown type: " + value );
+                    throw new IOException( "Unknown type: " + value );
             }
         }
 

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

@@ -100,12 +100,26 @@ public void pullAll( StreamCollector collector )
     }
 
     @Override
-    public void reset( StreamCollector collector )
+    public void reset()
     {
         try
         {
             markAsInUse();
-            delegate.reset( collector );
+            delegate.reset();
+        }
+        finally
+        {
+            markAsAvailable();
+        }
+    }
+
+    @Override
+    public void ackFailure()
+    {
+        try
+        {
+            markAsInUse();
+            delegate.ackFailure();
         }
         finally
         {
@@ -181,6 +195,12 @@ public void onError( Runnable runnable )
         delegate.onError( runnable );
     }
 
+    @Override
+    public boolean hasUnrecoverableErrors()
+    {
+        return delegate.hasUnrecoverableErrors();
+    }
+
     private void markAsAvailable()
     {
         inUse.set( false );

driver/src/main/java/org/neo4j/driver/internal/net/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/net/SocketClient.java

@@ -161,9 +161,7 @@ public void receiveOne( SocketResponseHandler handler ) throws IOException
     {
         reader.read( handler );
 
-        // TODO: all the errors come from the following trace should result in the termination of this channel
-        // https://github.com/neo4j/neo4j/blob/3
-        // .0/community/bolt/src/main/java/org/neo4j/bolt/v1/transport/BoltProtocolV1.java#L86
+        // Stop immediately if bolt protocol error happened on the server
         if ( handler.protocolViolationErrorOccurred() )
         {
             stop();

driver/src/main/java/org/neo4j/driver/internal/net/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.security.SecurityPlan;
 import org.neo4j.driver.internal.spi.Connection;
@@ -38,7 +37,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
 {
@@ -67,7 +68,7 @@ public SocketConnection( BoltServerAddress address, SecurityPlan securityPlan, L
     @Override
     public void init( String clientName, Map<String,Value> authToken )
     {
-        queueMessage( new InitMessage( clientName, authToken ), StreamCollector.NO_OP );
+        queueMessage( new InitMessage( clientName, authToken ), StreamCollector.INIT );
         sync();
     }
 
@@ -90,9 +91,15 @@ public void pullAll( StreamCollector collector )
     }
 
     @Override
-    public void reset( StreamCollector collector )
+    public void reset()
     {
-        queueMessage( ResetMessage.RESET, collector );
+        queueMessage( RESET, StreamCollector.RESET );
+    }
+
+    @Override
+    public void ackFailure()
+    {
+        queueMessage( ACK_FAILURE, StreamCollector.ACK_FAILURE );
     }
 
     @Override
@@ -147,7 +154,6 @@ private void assertNoServerFailure()
     {
         if ( responseHandler.serverFailureOccurred() )
         {
-            reset( StreamCollector.NO_OP );
             Neo4jException exception = responseHandler.serverFailure();
             responseHandler.clearError();
             throw exception;
@@ -194,4 +200,10 @@ public void onError( Runnable runnable )
     {
         throw new UnsupportedOperationException( "Error subscribers are not supported on SocketConnection." );
     }
+
+    @Override
+    public boolean hasUnrecoverableErrors()
+    {
+        throw new UnsupportedOperationException( "Unrecoverable error detection is not supported on SocketConnection." );
+    }
 }