Automatically close on session.reset()

pontusmelke · pontusmelke · commit dba1165cbbed · 2016-09-20T14:39:48.000+02:00
In the event of a reset the transaction should be completely closed and the
session should be ready to create new transactions without explicitly closing
any transactions.
diff --git a/driver/src/main/java/org/neo4j/driver/internal/NetworkSession.java b/driver/src/main/java/org/neo4j/driver/internal/NetworkSession.java
@@ -49,10 +49,13 @@ public class NetworkSession implements Session
         @Override
         public void run()
         {
-            if ( currentTransaction != null )
+            synchronized ( NetworkSession.this )
             {
-                lastBookmark = currentTransaction.bookmark();
-                currentTransaction = null;
+                if ( currentTransaction != null )
+                {
+                    lastBookmark = currentTransaction.bookmark();
+                    currentTransaction = null;
+                }
             }
         }
     };
@@ -73,9 +76,9 @@ public StatementResult run( String statementText )
     }
 
     @Override
-    public StatementResult run( String statementText, Map<String, Object> statementParameters )
+    public StatementResult run( String statementText, Map<String,Object> statementParameters )
     {
-        Value params = statementParameters == null ? Values.EmptyMap : value(statementParameters);
+        Value params = statementParameters == null ? Values.EmptyMap : value( statementParameters );
         return run( statementText, params );
     }
 
@@ -97,21 +100,24 @@ public StatementResult run( Statement statement )
     {
         ensureConnectionIsValidBeforeRunningSession();
         InternalStatementResult cursor = new InternalStatementResult( connection, null, statement );
-        connection.run( statement.text(), statement.parameters().asMap( Values.ofValue() ), cursor.runResponseCollector() );
+        connection.run( statement.text(), statement.parameters().asMap( Values.ofValue() ),
+                cursor.runResponseCollector() );
         connection.pullAll( cursor.pullAllResponseCollector() );
         connection.flush();
         return cursor;
     }
 
-    public void reset()
+    public synchronized void reset()
     {
         ensureSessionIsOpen();
         ensureNoUnrecoverableError();
         ensureConnectionIsOpen();
 
-        if( currentTransaction != null )
+        if ( currentTransaction != null )
         {
             currentTransaction.markToClose();
+            lastBookmark = currentTransaction.bookmark();
+            currentTransaction = null;
         }
         connection.resetAsync();
     }
@@ -126,21 +132,24 @@ public boolean isOpen()
     public void close()
     {
         // Use atomic operation to protect from closing the connection twice (putting back to the pool twice).
-        if( !isOpen.compareAndSet( true, false ) )
+        if ( !isOpen.compareAndSet( true, false ) )
         {
             throw new ClientException( "This session has already been closed." );
         }
         else
         {
-            if ( currentTransaction != null )
+            synchronized ( this )
             {
-                try
-                {
-                    currentTransaction.close();
-                }
-                catch ( Throwable e )
+                if ( currentTransaction != null )
                 {
-                    // Best-effort
+                    try
+                    {
+                        currentTransaction.close();
+                    }
+                    catch ( Throwable e )
+                    {
+                        // Best-effort
+                    }
                 }
             }
             try
@@ -167,7 +176,7 @@ public Transaction beginTransaction()
     }
 
     @Override
-    public Transaction beginTransaction( String bookmark )
+    public synchronized Transaction beginTransaction( String bookmark )
     {
         ensureConnectionIsValidBeforeOpeningTransaction();
         currentTransaction = new ExplicitTransaction( connection, txCleanup, bookmark );
@@ -224,7 +233,7 @@ private void ensureConnectionIsValidBeforeOpeningTransaction()
     @Override
     protected void finalize() throws Throwable
     {
-        if( isOpen.compareAndSet( true, false ) )
+        if ( isOpen.compareAndSet( true, false ) )
         {
             logger.error( "Neo4j Session object leaked, please ensure that your application calls the `close` " +
                           "method on Sessions before disposing of the objects.", null );
@@ -235,14 +244,15 @@ protected void finalize() throws Throwable
 
     private void ensureNoUnrecoverableError()
     {
-        if( connection.hasUnrecoverableErrors() )
+        if ( connection.hasUnrecoverableErrors() )
         {
             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." );
         }
     }
 
+    //should be called from a synchronized block
     private void ensureNoOpenTransactionBeforeRunningSession()
     {
         if ( currentTransaction != null )
@@ -252,6 +262,7 @@ private void ensureNoOpenTransactionBeforeRunningSession()
         }
     }
 
+    //should be called from a synchronized block
     private void ensureNoOpenTransactionBeforeOpeningTransaction()
     {
         if ( currentTransaction != null )
@@ -273,12 +284,13 @@ private void ensureConnectionIsOpen()
 
     private void ensureSessionIsOpen()
     {
-        if( !isOpen() )
+        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 " +
+                    "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." );
         }
     }
diff --git a/driver/src/test/java/org/neo4j/driver/v1/integration/TransactionIT.java b/driver/src/test/java/org/neo4j/driver/v1/integration/TransactionIT.java
@@ -23,6 +23,9 @@
 import org.junit.rules.ExpectedException;
 
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 import org.neo4j.driver.v1.Record;
 import org.neo4j.driver.v1.StatementResult;
@@ -142,7 +145,7 @@ public void shouldBeOpenBeforeCommit() throws Throwable
     public void shouldHandleNullParametersGracefully()
     {
         // When
-        session.run("match (n) return count(n)", (Value)null);
+        session.run( "match (n) return count(n)", (Value) null );
 
         // Then
         // pass - no exception thrown
@@ -155,7 +158,7 @@ public void shouldHandleFailureAfterClosingTransaction()
     {
         // GIVEN a successful query in a transaction
         Transaction tx = session.beginTransaction();
-        StatementResult result = tx.run("CREATE (n) RETURN n");
+        StatementResult result = tx.run( "CREATE (n) RETURN n" );
         result.consume();
         tx.success();
         tx.close();
@@ -164,7 +167,7 @@ public void shouldHandleFailureAfterClosingTransaction()
         exception.expect( ClientException.class );
 
         //WHEN running a malformed query in the original session
-        session.run("CREAT (n) RETURN n").consume();
+        session.run( "CREAT (n) RETURN n" ).consume();
     }
 
     @SuppressWarnings( "ConstantConditions" )
@@ -204,11 +207,102 @@ public void shouldHandleNullMapParameters() throws Throwable
         // When
         try ( Transaction tx = session.beginTransaction() )
         {
-            Map<String, Object> params = null;
+            Map<String,Object> params = null;
             tx.run( "CREATE (n:FirstNode)", params );
             tx.success();
         }
 
         // Then it wasn't the end of the world as we know it
     }
+
+    @Test
+    public void shouldHandleReset() throws Throwable
+    {
+            Transaction tx = session.beginTransaction();
+            tx.run( "UNWIND range(1,1) AS n RETURN n AS number" );
+            //TODO I believe there is a bug server side, if the RESET comes in
+            //too fast, stuff breaks
+            Thread.sleep( 100 );
+            session.reset();
+            tx = session.beginTransaction();
+            tx.run( "CREATE (n:FirstNode)" );
+            tx.success();
+            tx.close();
+
+            // Then the outcome of both statements should be visible
+            StatementResult result = session.run( "MATCH (n) RETURN count(n)" );
+            long nodes = result.single().get( "count(n)" ).asLong();
+            assertThat( nodes, equalTo( 1L ) );
+    }
+
+    @Test
+    public void shouldHandleResetBeforeRun() throws Throwable
+    {
+        // Expect
+        exception.expect( ClientException.class );
+        exception.expectMessage( "Cannot run more statements in this transaction, because previous statements in the " +
+                                 "transaction has failed and the transaction has been rolled back. Please start a new" +
+                                 " transaction to run another statement." );
+        // When
+        Transaction tx = session.beginTransaction();
+        session.reset();
+        tx.run( "CREATE (n:FirstNode)" );
+    }
+
+    private Transaction globalTx = null;
+    @Test
+    public void shouldHandleResetFromMultipleThreads() throws Throwable
+    {
+        // When
+        ExecutorService runner = Executors.newFixedThreadPool( 2 );
+        runner.execute( new Runnable()
+
+        {
+            @Override
+            public void run()
+            {
+                globalTx = session.beginTransaction();
+                    globalTx.run( "CREATE (n:FirstNode)" );
+                try
+                {
+                    Thread.sleep( 1000 );
+                }
+                catch ( InterruptedException e )
+                {
+                    new AssertionError( e );
+                }
+
+                globalTx = session.beginTransaction();
+                globalTx.run( "CREATE (n:FirstNode)" );
+                globalTx.success();
+                globalTx.close();
+
+            }
+        } );
+        runner.execute( new Runnable()
+
+        {
+            @Override
+            public void run()
+            {
+                try
+                {
+                    Thread.sleep( 500 );
+                }
+                catch ( InterruptedException e )
+                {
+                    new AssertionError( e );
+                }
+
+                session.reset();
+            }
+        } );
+
+        runner.awaitTermination( 5, TimeUnit.SECONDS );
+
+        // Then the outcome of both statements should be visible
+        StatementResult result = session.run( "MATCH (n) RETURN count(n)" );
+        long nodes = result.single().get( "count(n)" ).asLong();
+        assertThat( nodes, equalTo( 1L ) );
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -49,10 +49,13 @@ public class NetworkSession implements Session`
`49`	`49`	`@Override`
`50`	`50`	`public void run()`
`51`	`51`	`{`
`52`		`- if ( currentTransaction != null )`
	`52`	`+ synchronized ( NetworkSession.this )`
`53`	`53`	`{`
`54`		`- lastBookmark = currentTransaction.bookmark();`
`55`		`- currentTransaction = null;`
	`54`	`+ if ( currentTransaction != null )`
	`55`	`+ {`
	`56`	`+ lastBookmark = currentTransaction.bookmark();`
	`57`	`+ currentTransaction = null;`
	`58`	`+ }`
`56`	`59`	`}`
`57`	`60`	`}`
`58`	`61`	`};`
`@@ -73,9 +76,9 @@ public StatementResult run( String statementText )`
`73`	`76`	`}`
`74`	`77`
`75`	`78`	`@Override`
`76`		`- public StatementResult run( String statementText, Map<String, Object> statementParameters )`
	`79`	`+ public StatementResult run( String statementText, Map<String,Object> statementParameters )`
`77`	`80`	`{`
`78`		`- Value params = statementParameters == null ? Values.EmptyMap : value(statementParameters);`
	`81`	`+ Value params = statementParameters == null ? Values.EmptyMap : value( statementParameters );`
`79`	`82`	`return run( statementText, params );`
`80`	`83`	`}`
`81`	`84`
`@@ -97,21 +100,24 @@ public StatementResult run( Statement statement )`
`97`	`100`	`{`
`98`	`101`	`ensureConnectionIsValidBeforeRunningSession();`
`99`	`102`	`InternalStatementResult cursor = new InternalStatementResult( connection, null, statement );`
`100`		`- connection.run( statement.text(), statement.parameters().asMap( Values.ofValue() ), cursor.runResponseCollector() );`
	`103`	`+ connection.run( statement.text(), statement.parameters().asMap( Values.ofValue() ),`
	`104`	`+ cursor.runResponseCollector() );`
`101`	`105`	`connection.pullAll( cursor.pullAllResponseCollector() );`
`102`	`106`	`connection.flush();`
`103`	`107`	`return cursor;`
`104`	`108`	`}`
`105`	`109`
`106`		`- public void reset()`
	`110`	`+ public synchronized void reset()`
`107`	`111`	`{`
`108`	`112`	`ensureSessionIsOpen();`
`109`	`113`	`ensureNoUnrecoverableError();`
`110`	`114`	`ensureConnectionIsOpen();`
`111`	`115`
`112`		`- if( currentTransaction != null )`
	`116`	`+ if ( currentTransaction != null )`
`113`	`117`	`{`
`114`	`118`	`currentTransaction.markToClose();`
	`119`	`+ lastBookmark = currentTransaction.bookmark();`
	`120`	`+ currentTransaction = null;`
`115`	`121`	`}`
`116`	`122`	`connection.resetAsync();`
`117`	`123`	`}`
`@@ -126,21 +132,24 @@ public boolean isOpen()`
`126`	`132`	`public void close()`
`127`	`133`	`{`
`128`	`134`	`// Use atomic operation to protect from closing the connection twice (putting back to the pool twice).`
`129`		`- if( !isOpen.compareAndSet( true, false ) )`
	`135`	`+ if ( !isOpen.compareAndSet( true, false ) )`
`130`	`136`	`{`
`131`	`137`	`throw new ClientException( "This session has already been closed." );`
`132`	`138`	`}`
`133`	`139`	`else`
`134`	`140`	`{`
`135`		`- if ( currentTransaction != null )`
	`141`	`+ synchronized ( this )`
`136`	`142`	`{`
`137`		`- try`
`138`		`- {`
`139`		`- currentTransaction.close();`
`140`		`- }`
`141`		`- catch ( Throwable e )`
	`143`	`+ if ( currentTransaction != null )`
`142`	`144`	`{`
`143`		`- // Best-effort`
	`145`	`+ try`
	`146`	`+ {`
	`147`	`+ currentTransaction.close();`
	`148`	`+ }`
	`149`	`+ catch ( Throwable e )`
	`150`	`+ {`
	`151`	`+ // Best-effort`
	`152`	`+ }`
`144`	`153`	`}`
`145`	`154`	`}`
`146`	`155`	`try`
`@@ -167,7 +176,7 @@ public Transaction beginTransaction()`
`167`	`176`	`}`
`168`	`177`
`169`	`178`	`@Override`
`170`		`- public Transaction beginTransaction( String bookmark )`
	`179`	`+ public synchronized Transaction beginTransaction( String bookmark )`
`171`	`180`	`{`
`172`	`181`	`ensureConnectionIsValidBeforeOpeningTransaction();`
`173`	`182`	`currentTransaction = new ExplicitTransaction( connection, txCleanup, bookmark );`
`@@ -224,7 +233,7 @@ private void ensureConnectionIsValidBeforeOpeningTransaction()`
`224`	`233`	`@Override`
`225`	`234`	`protected void finalize() throws Throwable`
`226`	`235`	`{`
`227`		`- if( isOpen.compareAndSet( true, false ) )`
	`236`	`+ if ( isOpen.compareAndSet( true, false ) )`
`228`	`237`	`{`
`229`	`238`	logger.error( "Neo4j Session object leaked, please ensure that your application calls the `close` " +
`230`	`239`	`"method on Sessions before disposing of the objects.", null );`
`@@ -235,14 +244,15 @@ protected void finalize() throws Throwable`
`235`	`244`
`236`	`245`	`private void ensureNoUnrecoverableError()`
`237`	`246`	`{`
`238`		`- if( connection.hasUnrecoverableErrors() )`
	`247`	`+ if ( connection.hasUnrecoverableErrors() )`
`239`	`248`	`{`
`240`	`249`	`throw new ClientException( "Cannot run more statements in the current session as an unrecoverable error " +`
`241`	`250`	`"has happened. Please close the current session and re-run your statement in a" +`
`242`	`251`	`" new session." );`
`243`	`252`	`}`
`244`	`253`	`}`
`245`	`254`
	`255`	`+ //should be called from a synchronized block`
`246`	`256`	`private void ensureNoOpenTransactionBeforeRunningSession()`
`247`	`257`	`{`
`248`	`258`	`if ( currentTransaction != null )`
`@@ -252,6 +262,7 @@ private void ensureNoOpenTransactionBeforeRunningSession()`
`252`	`262`	`}`
`253`	`263`	`}`
`254`	`264`
	`265`	`+ //should be called from a synchronized block`
`255`	`266`	`private void ensureNoOpenTransactionBeforeOpeningTransaction()`
`256`	`267`	`{`
`257`	`268`	`if ( currentTransaction != null )`
`@@ -273,12 +284,13 @@ private void ensureConnectionIsOpen()`
`273`	`284`
`274`	`285`	`private void ensureSessionIsOpen()`
`275`	`286`	`{`
`276`		`- if( !isOpen() )`
	`287`	`+ if ( !isOpen() )`
`277`	`288`	`{`
`278`	`289`	`throw new ClientException(`
`279`	`290`	`"No more interaction with this session is allowed " +`
`280`	`291`	`"as the current session is already closed or marked as closed. " +`
`281`		`- "You get this error either because you have a bad reference to a session that has already be closed " +`
	`292`	`+ "You get this error either because you have a bad reference to a session that has already be " +`
	`293`	`+ "closed " +`
`282`	`294`	"or you are trying to reuse a session that you have called `reset` on it." );
`283`	`295`	`}`
`284`	`296`	`}`