Sync only open connections

Added a check to sync only open connections before closing them in the session.

Author: lutovich

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

@@ -91,7 +91,7 @@ public StatementResult run( Statement statement )
         ensureSessionIsOpen();
         ensureNoOpenTransactionBeforeRunningSession();
 
-        closeCurrentConnection( true );
+        syncAndCloseCurrentConnection();
         currentConnection = acquireConnection();
 
         return run( currentConnection, statement, this );
@@ -140,13 +140,6 @@ public void close()
             throw new ClientException( "This session has already been closed." );
         }
 
-        if ( currentConnection != null && !currentConnection.isOpen() )
-        {
-            // the socket connection is already closed due to some error, cannot send more data
-            closeCurrentConnection( false );
-            return;
-        }
-
         synchronized ( this )
         {
             if ( currentTransaction != null )
@@ -162,7 +155,8 @@ public void close()
                 }
             }
         }
-        closeCurrentConnection( true );
+        
+        syncAndCloseCurrentConnection();
     }
 
     @Override
@@ -177,7 +171,7 @@ public synchronized Transaction beginTransaction( String bookmark )
         ensureSessionIsOpen();
         ensureNoOpenTransactionBeforeOpeningTransaction();
 
-        closeCurrentConnection( true );
+        syncAndCloseCurrentConnection();
         currentConnection = acquireConnection();
 
         currentTransaction = new ExplicitTransaction( currentConnection, this, bookmark );
@@ -200,15 +194,15 @@ public TypeSystem typeSystem()
     @Override
     public synchronized void onResultConsumed()
     {
-        closeCurrentConnection( false );
+        closeCurrentConnection();
     }
 
     @Override
     public synchronized void onTransactionClosed( ExplicitTransaction tx )
     {
         if ( currentTransaction != null && currentTransaction == tx )
         {
-            closeCurrentConnection( false );
+            closeCurrentConnection();
             lastBookmark = currentTransaction.bookmark();
             currentTransaction = null;
         }
@@ -286,6 +280,16 @@ boolean currentConnectionIsOpen()
         return currentConnection != null && currentConnection.isOpen();
     }
 
+    private void syncAndCloseCurrentConnection()
+    {
+        closeCurrentConnection( true );
+    }
+
+    private void closeCurrentConnection()
+    {
+        closeCurrentConnection( false );
+    }
+
     private void closeCurrentConnection( boolean sync )
     {
         if ( currentConnection == null )
@@ -297,7 +301,7 @@ private void closeCurrentConnection( boolean sync )
         currentConnection = null;
         try
         {
-            if ( sync )
+            if ( sync && connection.isOpen() )
             {
                 connection.sync();
             }

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

@@ -22,6 +22,7 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.mockito.InOrder;
 
 import java.util.Map;
 
@@ -47,6 +48,7 @@
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.RETURNS_MOCKS;
 import static org.mockito.Mockito.anyMapOf;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -204,11 +206,13 @@ public void acquiresNewConnectionForRun()
     }
 
     @Test
-    public void closesPreviousConnectionForRun()
+    public void syncsAndClosesPreviousConnectionForRun()
     {
         ConnectionProvider connectionProvider = mock( ConnectionProvider.class );
         PooledConnection connection1 = mock( PooledConnection.class );
+        when( connection1.isOpen() ).thenReturn( true );
         PooledConnection connection2 = mock( PooledConnection.class );
+        when( connection2.isOpen() ).thenReturn( true );
         when( connectionProvider.acquireConnection( READ ) ).thenReturn( connection1 ).thenReturn( connection2 );
         NetworkSession session = newSession( connectionProvider, READ );
 
@@ -220,7 +224,29 @@ public void closesPreviousConnectionForRun()
         verify( connectionProvider, times( 2 ) ).acquireConnection( READ );
         verify( connection2 ).run( eq( "RETURN 2" ), anyParams(), any( Collector.class ) );
 
-        verify( connection1 ).sync();
+        InOrder inOrder = inOrder( connection1 );
+        inOrder.verify( connection1 ).sync();
+        inOrder.verify( connection1 ).close();
+    }
+
+    @Test
+    public void closesPreviousBrokenConnectionForRun()
+    {
+        ConnectionProvider connectionProvider = mock( ConnectionProvider.class );
+        PooledConnection connection1 = mock( PooledConnection.class );
+        PooledConnection connection2 = mock( PooledConnection.class );
+        when( connectionProvider.acquireConnection( READ ) ).thenReturn( connection1 ).thenReturn( connection2 );
+        NetworkSession session = newSession( connectionProvider, READ );
+
+        session.run( "RETURN 1" );
+        verify( connectionProvider ).acquireConnection( READ );
+        verify( connection1 ).run( eq( "RETURN 1" ), anyParams(), any( Collector.class ) );
+
+        session.run( "RETURN 2" );
+        verify( connectionProvider, times( 2 ) ).acquireConnection( READ );
+        verify( connection2 ).run( eq( "RETURN 2" ), anyParams(), any( Collector.class ) );
+
+        verify( connection1, never() ).sync();
         verify( connection1 ).close();
     }