Make sessioin.reset thread-safe

Author: Zhen

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

@@ -59,6 +59,7 @@ public void run()
 
     private ExplicitTransaction currentTransaction;
     private AtomicBoolean isOpen = new AtomicBoolean( true );
+    private AtomicBoolean markedToClose = new AtomicBoolean( false );
 
     NetworkSession( Connection connection, Logger logger )
     {
@@ -108,13 +109,16 @@ public void reset()
         ensureNoUnrecoverableError();
         ensureConnectionIsOpen();
 
-        connection.resetAsync();
+        if( markedToClose.compareAndSet( false, true ) )
+        {
+            connection.resetAsync();
+        }
     }
 
     @Override
     public boolean isOpen()
     {
-        return isOpen.get();
+        return isOpen.get() && !markedToClose.get();
     }
 
     @Override
@@ -263,4 +267,16 @@ private void ensureConnectionIsOpen()
                                        "Please close this session and retry your statement in another new session." );
         }
     }
+
+    private void ensureSessionIsOpen()
+    {
+        if( !isOpen() )
+        {
+            throw new ClientException(
+                    "No more interaction with this session is allowed " +
+                    "as the current session is already closed or marked as closed. " +
+                    "You get this error either because you have a bad reference to a session that has already be closed " +
+                    "or you are trying to reuse a session that you have called `reset` on it." );
+        }
+    }
 }

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

@@ -113,7 +113,7 @@ public void sync()
     }
 
     @Override
-    public void flush()
+    public synchronized void flush()
     {
         try
         {
@@ -180,7 +180,7 @@ else if ( e instanceof SocketTimeoutException )
         }
     }
 
-    private void queueMessage( Message msg, Collector collector )
+    private synchronized void queueMessage( Message msg, Collector collector )
     {
         pendingMessages.add( msg );
         responseHandler.appendResultCollector( collector );

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

@@ -18,9 +18,9 @@
  */
 package org.neo4j.driver.internal.net;
 
-import java.util.LinkedList;
 import java.util.Map;
 import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 import org.neo4j.driver.internal.messaging.MessageHandler;
 import org.neo4j.driver.internal.spi.Collector;
@@ -39,7 +39,7 @@
 
 public class SocketResponseHandler implements MessageHandler
 {
-    private final Queue<Collector> collectors = new LinkedList<>();
+    private final Queue<Collector> collectors = new ConcurrentLinkedQueue<>();
 
     /** If a failure occurs, the error gets stored here */
     private Neo4jException error;

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

@@ -105,7 +105,8 @@ public void shouldKillLongRunningStatement() throws Throwable
         final int killTimeout = 1; // 1s
         long startTime = -1, endTime;
 
-        try( final Session session = driver.session() )
+        final Session session = driver.session();
+        try
         {
             StatementResult result =
                     session.run( "CALL test.driver.longRunningStatement({seconds})",
@@ -123,13 +124,18 @@ public void shouldKillLongRunningStatement() throws Throwable
         {
             endTime = System.currentTimeMillis();
             assertTrue( startTime > 0 );
-            assertTrue( endTime - startTime > killTimeout * 1000 ); // get killed by session.kill
+            assertTrue( endTime - startTime > killTimeout * 1000 ); // get reset by session.reset
             assertTrue( endTime - startTime < executionTimeout * 1000 / 2 ); // finished before execution finished
+            assertFalse( session.isOpen() );
         }
         catch ( Exception e )
         {
             fail( "Should be a Neo4jException" );
         }
+        finally
+        {
+            session.close();
+        }
     }
 
     @Test
@@ -168,7 +174,7 @@ public void shouldKillLongStreamingResult() throws Throwable
             assertThat( recordCount, greaterThan(1) );
 
             assertTrue( startTime > 0 );
-            assertTrue( endTime - startTime > killTimeout * 1000 ); // get killed by session.kill
+            assertTrue( endTime - startTime > killTimeout * 1000 ); // get reset by session.reset
             assertTrue( endTime - startTime < executionTimeout * 1000 / 2 ); // finished before execution finished
         }
     }
@@ -190,7 +196,7 @@ public void run()
                 }
                 finally
                 {
-                    session.reset(); // kill the session after timeout
+                    session.reset(); // reset the session after timeout
                 }
             }
         } ).start();