Merge pull request #231 from pontusmelke/1.1-close-tx-on-reset

nigelsmall · web-flow · commit d033d8b05f38 · 2016-09-22T14:48:58.000+01:00
Automatically close on session.reset()
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/main/java/org/neo4j/driver/internal/net/ConcurrencyGuardingConnection.java b/driver/src/main/java/org/neo4j/driver/internal/net/ConcurrencyGuardingConnection.java
@@ -208,9 +208,9 @@ public void resetAsync()
     }
 
     @Override
-    public boolean isInterrupted()
+    public boolean isAckFailureMuted()
     {
-        return delegate.isInterrupted();
+        return delegate.isAckFailureMuted();
     }
 
     private void markAsAvailable()
diff --git a/driver/src/main/java/org/neo4j/driver/internal/net/SocketConnection.java b/driver/src/main/java/org/neo4j/driver/internal/net/SocketConnection.java
@@ -46,7 +46,8 @@ public class SocketConnection implements Connection
 {
     private final Queue<Message> pendingMessages = new LinkedList<>();
     private final SocketResponseHandler responseHandler;
-    private AtomicBoolean interrupted = new AtomicBoolean( false );
+    private AtomicBoolean isInterrupted = new AtomicBoolean( false );
+    private AtomicBoolean isAckFailureMuted = new AtomicBoolean( false );
     private final Collector.InitCollector initCollector = new Collector.InitCollector();
 
     private final SocketClient socket;
@@ -115,6 +116,8 @@ public void sync()
     @Override
     public synchronized void flush()
     {
+        ensureNotInterrupted();
+
         try
         {
             socket.send( pendingMessages );
@@ -126,6 +129,29 @@ public synchronized void flush()
         }
     }
 
+    private void ensureNotInterrupted()
+    {
+        try
+        {
+            if( isInterrupted.get() )
+            {
+                // receive each of it and throw error immediately
+                while ( responseHandler.collectorsWaiting() > 0 )
+                {
+                    receiveOne();
+                }
+            }
+        }
+        catch ( Neo4jException e )
+        {
+            throw new ClientException(
+                    "An error has occurred due to the cancellation of executing a previous statement. " +
+                    "You received this error probably because you did not consume the result immediately after " +
+                    "running the statement which get reset in this session.", e );
+        }
+
+    }
+
     private void receiveAll()
     {
         try
@@ -159,6 +185,7 @@ private void assertNoServerFailure()
         {
             Neo4jException exception = responseHandler.serverFailure();
             responseHandler.clearError();
+            isInterrupted.set( false );
             throw exception;
         }
     }
@@ -182,6 +209,8 @@ else if ( e instanceof SocketTimeoutException )
 
     private synchronized void queueMessage( Message msg, Collector collector )
     {
+        ensureNotInterrupted();
+
         pendingMessages.add( msg );
         responseHandler.appendResultCollector( collector );
     }
@@ -211,26 +240,26 @@ public boolean hasUnrecoverableErrors()
     }
 
     @Override
-    public void resetAsync()
+    public synchronized void resetAsync()
     {
-        if( interrupted.compareAndSet( false, true ) )
+        queueMessage( RESET, new Collector.ResetCollector( new Runnable()
         {
-            queueMessage( RESET, new Collector.ResetCollector( new Runnable()
+            @Override
+            public void run()
             {
-                @Override
-                public void run()
-                {
-                    interrupted.set( false );
-                }
-            } ) );
-            flush();
-        }
+                isInterrupted.set( false );
+                isAckFailureMuted.set( false );
+            }
+        } ) );
+        flush();
+        isInterrupted.set( true );
+        isAckFailureMuted.set( true );
     }
 
     @Override
-    public boolean isInterrupted()
+    public boolean isAckFailureMuted()
     {
-        return interrupted.get();
+        return isAckFailureMuted.get();
     }
 
     @Override
diff --git a/driver/src/main/java/org/neo4j/driver/internal/net/pooling/PooledConnection.java b/driver/src/main/java/org/neo4j/driver/internal/net/pooling/PooledConnection.java
@@ -226,9 +226,9 @@ public void resetAsync()
     }
 
     @Override
-    public boolean isInterrupted()
+    public boolean isAckFailureMuted()
     {
-        return delegate.isInterrupted();
+        return delegate.isAckFailureMuted();
     }
 
     @Override
@@ -260,7 +260,7 @@ private void onDelegateException( RuntimeException e )
         {
             unrecoverableErrorsOccurred = true;
         }
-        else if( !isInterrupted() )
+        else if( !isAckFailureMuted() )
         {
             ackFailure();
         }
diff --git a/driver/src/main/java/org/neo4j/driver/internal/spi/Connection.java b/driver/src/main/java/org/neo4j/driver/internal/spi/Connection.java
@@ -114,10 +114,10 @@ public interface Connection extends AutoCloseable
     void resetAsync();
 
     /**
-     * Return true if the current session statement execution has been interrupted by another thread, otherwise false.
-     * @return true if the current session statement execution has been interrupted by another thread, otherwise false
+     * Return true if ack_failure message is temporarily muted as the failure message will be acked using reset instead
+     * @return true if no ack_failre message should be sent when ackable failures are received.
      */
-    boolean isInterrupted();
+    boolean isAckFailureMuted();
 
     /**
      * Returns the version of the server connected to.
diff --git a/driver/src/test/java/org/neo4j/driver/v1/integration/SessionIT.java b/driver/src/test/java/org/neo4j/driver/v1/integration/SessionIT.java
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

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`	`}`
Original file line number	Diff line number	Diff line change
`@@ -208,9 +208,9 @@ public void resetAsync()`
`208`	`208`	`}`
`209`	`209`
`210`	`210`	`@Override`
`211`		`- public boolean isInterrupted()`
	`211`	`+ public boolean isAckFailureMuted()`
`212`	`212`	`{`
`213`		`- return delegate.isInterrupted();`
	`213`	`+ return delegate.isAckFailureMuted();`
`214`	`214`	`}`
`215`	`215`
`216`	`216`	`private void markAsAvailable()`
Original file line number	Diff line number	Diff line change
`@@ -226,9 +226,9 @@ public void resetAsync()`
`226`	`226`	`}`
`227`	`227`
`228`	`228`	`@Override`
`229`		`- public boolean isInterrupted()`
	`229`	`+ public boolean isAckFailureMuted()`
`230`	`230`	`{`
`231`		`- return delegate.isInterrupted();`
	`231`	`+ return delegate.isAckFailureMuted();`
`232`	`232`	`}`
`233`	`233`
`234`	`234`	`@Override`
`@@ -260,7 +260,7 @@ private void onDelegateException( RuntimeException e )`
`260`	`260`	`{`
`261`	`261`	`unrecoverableErrorsOccurred = true;`
`262`	`262`	`}`
`263`		`- else if( !isInterrupted() )`
	`263`	`+ else if( !isAckFailureMuted() )`
`264`	`264`	`{`
`265`	`265`	`ackFailure();`
`266`	`266`	`}`