Mark a few [client message, server response messages] pairs as invalid message response pairs,

these message pairs are listed as follows:

INIT        <- IGNORED
ACK_FAILURE <- IGNORED
ACK_FAILURE <- FAILURE
RESET       <- IGNORED
RESET       <- FAILURE

In case of seeing such message pairs, raise an unrecoverable error message on client side.

Author: Zhen

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

@@ -198,8 +198,8 @@ private void ensureNoUnrecoverableError()
     {
         if( connection.hasUnrecoverableErrors() )
         {
-            throw new ClientException( "Cannot run more statements in the current session as unrecoverable errors " +
-                                       "has happened. Please close the currect session and re-run your statement in a" +
+            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." );
         }
     }

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/connector/ConcurrencyGuardingConnection.java

@@ -100,12 +100,12 @@ public void pullAll( StreamCollector collector )
     }
 
     @Override
-    public void reset( StreamCollector collector )
+    public void reset()
     {
         try
         {
             markAsInUse();
-            delegate.reset( collector );
+            delegate.reset();
         }
         finally
         {
@@ -114,12 +114,12 @@ public void reset( StreamCollector collector )
     }
 
     @Override
-    public void ackFailure( StreamCollector collector )
+    public void ackFailure()
     {
         try
         {
             markAsInUse();
-            delegate.ackFailure( collector );
+            delegate.ackFailure();
         }
         finally
         {

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

@@ -129,9 +129,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/connector/socket/SocketConnection.java

@@ -67,7 +67,7 @@ public SocketConnection( String host, int port, Config config )
     @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,15 +90,15 @@ public void pullAll( StreamCollector collector )
     }
 
     @Override
-    public void reset( StreamCollector collector )
+    public void reset()
     {
-        queueMessage( RESET, collector );
+        queueMessage( RESET, StreamCollector.RESET );
     }
 
     @Override
-    public void ackFailure( StreamCollector collector )
+    public void ackFailure()
     {
-        queueMessage( ACK_FAILURE, collector );
+        queueMessage( ACK_FAILURE, StreamCollector.ACK_FAILURE );
     }
 
     @Override

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

@@ -76,7 +76,7 @@ public void handleFailureMessage( String code, String message )
         }
         if ( collector != null )
         {
-            collector.done();
+            collector.doneFailure( error );
         }
     }
 
@@ -90,7 +90,7 @@ public void handleSuccessMessage( Map<String,Value> meta )
         collectPlan( collector, meta.get( "plan" ) );
         collectProfile( collector, meta.get( "profile" ) );
         collectNotifications( collector, meta.get( "notifications" ) );
-        collector.done();
+        collector.doneSuccess();
     }
 
     private void collectNotifications( StreamCollector collector, Value notifications )
@@ -178,7 +178,7 @@ public void handleIgnoredMessage()
         StreamCollector collector = collectors.remove();
         if (collector != null)
         {
-            collector.done();
+            collector.doneIgnored();
         }
     }
 

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

@@ -327,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/pool/PooledConnection.java

@@ -124,11 +124,11 @@ public void pullAll( StreamCollector collector )
     }
 
     @Override
-    public void reset( StreamCollector collector )
+    public void reset()
     {
         try
         {
-            delegate.reset( collector );
+            delegate.reset();
         }
         catch ( RuntimeException e )
         {
@@ -137,11 +137,11 @@ public void reset( StreamCollector collector )
     }
 
     @Override
-    public void ackFailure( StreamCollector collector )
+    public void ackFailure()
     {
         try
         {
-            delegate.ackFailure( collector );
+            delegate.ackFailure();
         }
         catch ( RuntimeException e )
         {
@@ -230,7 +230,7 @@ private void onDelegateException( RuntimeException e )
         }
         else
         {
-            ackFailure( StreamCollector.NO_OP );
+            ackFailure();
         }
         if( onError != null )
         {

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

@@ -88,7 +88,7 @@ else if ( driverStopped.get() )
     boolean validConnection( PooledConnection pooledConnection )
     {
         // once the pooledConn has marked to have unrecoverable errors, there is no way to remove the error
-        // and we should close the conn without bother to reset the conn at all
+        // and we should close the conn without bothering to reset the conn at all
         return !pooledConnection.hasUnrecoverableErrors() &&
                reset(pooledConnection) &&
                (pooledConnection.idleTime() <= minIdleBeforeConnectionTest || ping( pooledConnection ));
@@ -104,7 +104,7 @@ private boolean reset( PooledConnection conn )
     {
         try
         {
-            conn.reset( StreamCollector.NO_OP );
+            conn.reset();
             conn.sync();
             return true;
         }

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

@@ -55,16 +55,17 @@ public interface Connection extends AutoCloseable
     void pullAll( StreamCollector collector );
 
     /**
-     * Queue a reset action, output will be handed to the collector once the pull starts. This will
+     * Queue a reset action, throw {@link org.neo4j.driver.v1.exceptions.ClientException} if an ignored message is received. This will
      * close the stream once its completed, allowing another {@link #run(String, java.util.Map, StreamCollector) run}
      */
-    void reset( StreamCollector collector );
+    void reset();
 
     /**
-     * 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}
+     * Queue a ack_failure action, valid output could only be success. Throw {@link org.neo4j.driver.v1.exceptions.ClientException} if
+     * a failure or ignored message is received. This will close the stream once it is completed, allowing another
+     * {@link #run(String, java.util.Map, StreamCollector) run}
      */
-    void ackFailure( StreamCollector collector );
+    void ackFailure();
 
     /**
      * Ensure all outstanding actions are carried out on the server.