Make sessioin.reset thread-safe

Sync on `enqueue` and `flush` to avoid sending `reset` and other messages at the same time.
Also changed response handler stream collector queue to be concurrent queue so that adding into the queue (sending) and removing from the queue (receiving) can be done safely concurrently.
Keep `ack_fail` muted when `reset` is already send but not replied yet.

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();