Close connection after result reading failure

When result fetching fails (cypher runtime error or network error) exception
is thrown but it is still possible to maintain a reference to the result
object. It'll just be empty after failure. It is not possible to fetch any
other records after failure and the underlying pooled connection can be
closed and returned to the pool.

This commit makes result return its connection to the pool on failure. This
connection (if healthy) can be used by subsequent sessions. Also added tests
for connection handling.

Author: lutovich

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

@@ -45,7 +45,6 @@
 
 import static java.util.Collections.emptyList;
 
-// todo: connection should be released if fetching from network fails, no need to sync...
 public class InternalStatementResult implements StatementResult
 {
     private final Connection connection;
@@ -289,7 +288,7 @@ public ResultSummary consume()
         {
             do
             {
-                connection.receiveOne();
+                receiveOne();
                 recordBuffer.clear();
             }
             while ( !done );
@@ -303,7 +302,7 @@ public ResultSummary summary()
     {
         while( !done )
         {
-            connection.receiveOne();
+            receiveOne();
         }
 
         return summary;
@@ -321,12 +320,28 @@ private boolean tryFetchNext()
         {
             if ( done )
             {
-                resourcesHandler.onResultConsumed();
                 return false;
             }
-            connection.receiveOne();
+            receiveOne();
         }
 
         return true;
     }
+
+    private void receiveOne()
+    {
+        try
+        {
+            connection.receiveOne();
+        }
+        catch ( Throwable error )
+        {
+            resourcesHandler.onResultConsumed();
+            throw error;
+        }
+        if ( done )
+        {
+            resourcesHandler.onResultConsumed();
+        }
+    }
 }

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

@@ -99,12 +99,12 @@ public StatementResult run( Statement statement )
 
     public static StatementResult run( Connection connection, Statement statement, SessionResourcesHandler resourcesHandler )
     {
-        InternalStatementResult cursor = new InternalStatementResult( connection, resourcesHandler, null, statement );
+        InternalStatementResult result = new InternalStatementResult( connection, resourcesHandler, null, statement );
         connection.run( statement.text(), statement.parameters().asMap( Values.ofValue() ),
-                cursor.runResponseCollector() );
-        connection.pullAll( cursor.pullAllResponseCollector() );
+                result.runResponseCollector() );
+        connection.pullAll( result.pullAllResponseCollector() );
         connection.flush();
-        return cursor;
+        return result;
     }
 
     @Override

driver/src/test/java/org/neo4j/driver/internal/InternalStatementResultTest.java

@@ -41,6 +41,7 @@
 import static java.util.Arrays.asList;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.collection.IsCollectionWithSize.hasSize;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
@@ -49,6 +50,7 @@
 import static org.junit.Assert.fail;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
 import static org.neo4j.driver.v1.Records.column;
 import static org.neo4j.driver.v1.Values.ofString;
 import static org.neo4j.driver.v1.Values.value;
@@ -384,12 +386,77 @@ public void shouldNotPeekIntoTheFutureWhenResultIsEmpty()
         Record future = result.peek();
     }
 
+    @Test
+    public void shouldNotifyResourcesHandlerWhenFetchedViaList()
+    {
+        SessionResourcesHandler resourcesHandler = mock( SessionResourcesHandler.class );
+        StatementResult result = createResult( 10, resourcesHandler );
+
+        List<Record> records = result.list();
+        assertEquals( 10, records.size() );
+
+        verify( resourcesHandler ).onResultConsumed();
+    }
+
+    @Test
+    public void shouldNotifyResourcesHandlerWhenFetchedViaSingle()
+    {
+        SessionResourcesHandler resourcesHandler = mock( SessionResourcesHandler.class );
+        StatementResult result = createResult( 1, resourcesHandler );
+
+        Record record = result.single();
+        assertEquals( "v1-1", record.get( "k1" ).asString() );
+
+        verify( resourcesHandler ).onResultConsumed();
+    }
+
+    @Test
+    public void shouldNotifyResourcesHandlerWhenFetchedViaIterator()
+    {
+        SessionResourcesHandler resourcesHandler = mock( SessionResourcesHandler.class );
+        StatementResult result = createResult( 1, resourcesHandler );
+
+        while ( result.hasNext() )
+        {
+            assertNotNull( result.next() );
+        }
+
+        verify( resourcesHandler ).onResultConsumed();
+    }
+
+    @Test
+    public void shouldNotifyResourcesHandlerWhenSummary()
+    {
+        SessionResourcesHandler resourcesHandler = mock( SessionResourcesHandler.class );
+        StatementResult result = createResult( 10, resourcesHandler );
+
+        assertNotNull( result.summary() );
+
+        verify( resourcesHandler ).onResultConsumed();
+    }
+
+    @Test
+    public void shouldNotifyResourcesHandlerWhenConsumed()
+    {
+        SessionResourcesHandler resourcesHandler = mock( SessionResourcesHandler.class );
+        StatementResult result = createResult( 5, resourcesHandler );
+
+        result.consume();
+
+        verify( resourcesHandler ).onResultConsumed();
+    }
+
     private StatementResult createResult( int numberOfRecords )
+    {
+        return createResult( numberOfRecords, SessionResourcesHandler.NO_OP );
+    }
+
+    private StatementResult createResult( int numberOfRecords, SessionResourcesHandler resourcesHandler )
     {
         Connection connection = mock( Connection.class );
         String statement = "<unknown>";
 
-        final InternalStatementResult result = new InternalStatementResult( connection, SessionResourcesHandler.NO_OP, null,
+        final InternalStatementResult result = new InternalStatementResult( connection, resourcesHandler, null,
                 new Statement( statement ) );
 
         // Each time the cursor calls `recieveOne`, we'll run one of these,