#80 fix handling of nested transactions

Author: aburmeis

src/main/java/com/arangodb/springframework/repository/query/QueryTransactionBridge.java

@@ -29,21 +29,41 @@
  * Bridge to postpone late transaction start to be able to inject collections from query side.
  */
 public class QueryTransactionBridge {
-    private static final Function<Collection<String>, String> NO_TRANSACTION = any -> null;
-    private static final ThreadLocal<Function<Collection<String>, String>> CURRENT_TRANSACTION = new NamedInheritableThreadLocal<>("ArangoTransactionBegin");
 
-    public QueryTransactionBridge() {
-        CURRENT_TRANSACTION.set(NO_TRANSACTION);
-    }
+    private static final ThreadLocal<Function<Collection<String>, String>> CURRENT_TRANSACTION = new NamedInheritableThreadLocal<Function<Collection<String>, String>>("ArangoTransactionBegin") {
+        @Override
+        protected Function<Collection<String>, String> initialValue() {
+            return any -> null;
+        }
+    };
 
+    /**
+     * Prepare the bridge for accepting transaction begin.
+     * @param begin a function accepting collection names and returning a stream transaction id
+     *
+     * @see com.arangodb.springframework.transaction.ArangoTransactionManager
+     */
     public void setCurrentTransaction(Function<Collection<String>, String> begin) {
         CURRENT_TRANSACTION.set(begin);
     }
 
+    /**
+     * Reset the bridge ignoring transaction begin.
+     *
+     * @see com.arangodb.springframework.transaction.ArangoTransactionManager
+     */
     public void clearCurrentTransaction() {
-        CURRENT_TRANSACTION.set(NO_TRANSACTION);
+        CURRENT_TRANSACTION.remove();
     }
 
+    /**
+     * Applies the collection names to any current transaction.
+     * @param collections additional collection names
+     * @return the stream transaction id or {@code null} without transaction
+     *
+     * @see AbstractArangoQuery
+     * @see com.arangodb.springframework.repository.SimpleArangoRepository
+     */
     public String getCurrentTransaction(Collection<String> collections) {
         return CURRENT_TRANSACTION.get().apply(collections);
     }

src/main/java/com/arangodb/springframework/transaction/ArangoTransactionResource.java renamed to src/main/java/com/arangodb/springframework/transaction/ArangoTransactionHolder.java

@@ -20,9 +20,12 @@
 
 package com.arangodb.springframework.transaction;
 
+import com.arangodb.entity.StreamTransactionEntity;
+import com.arangodb.entity.StreamTransactionStatus;
 import org.springframework.lang.Nullable;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.Set;
 
@@ -32,53 +35,41 @@
  * @see TransactionSynchronizationManager#bindResource(Object, Object)
  * @see ArangoTransactionObject
  */
-class ArangoTransactionResource {
+class ArangoTransactionHolder {
 
-    private String streamTransactionId;
-    private Set<String> collectionNames;
-
-    private boolean rollbackOnly;
-    int references = 0;
-
-    ArangoTransactionResource(@Nullable String streamTransactionId, Set<String> collectionNames, boolean rollbackOnly) {
-        this.streamTransactionId = streamTransactionId;
-        setCollectionNames(collectionNames);
-        this.rollbackOnly = rollbackOnly;
-    }
+    private final Set<String> collectionNames = new HashSet<>();
+    private StreamTransactionEntity transaction = null;
+    private boolean rollbackOnly = false;
 
+    @Nullable
     String getStreamTransactionId() {
-        return streamTransactionId;
+        return transaction == null ? null : transaction.getId();
     }
 
-    void setStreamTransactionId(String streamTransactionId) {
-        this.streamTransactionId = streamTransactionId;
+    void setStreamTransaction(StreamTransactionEntity transaction) {
+        this.transaction = transaction;
     }
 
     Set<String> getCollectionNames() {
         return collectionNames;
     }
 
-    void setCollectionNames(Set<String> collectionNames) {
-        this.collectionNames = new HashSet<>(collectionNames);
+    void addCollectionNames(Collection<String> collectionNames) {
+        if (transaction != null) {
+            throw new IllegalStateException("Collections must not be added after stream transaction begun");
+        }
+        this.collectionNames.addAll(collectionNames);
     }
 
     boolean isRollbackOnly() {
-        return rollbackOnly;
-    }
-
-    void setRollbackOnly(boolean rollbackOnly) {
-        this.rollbackOnly = rollbackOnly;
-    }
-
-    void increaseReferences() {
-        ++references;
+        return rollbackOnly || isStatus(StreamTransactionStatus.aborted);
     }
 
-    boolean isSingleReference() {
-        return references <= 1;
+    void setRollbackOnly() {
+        rollbackOnly = true;
     }
 
-    void decreasedReferences() {
-        --references;
+    public boolean isStatus(StreamTransactionStatus status) {
+        return transaction != null && transaction.getStatus() == status;
     }
 }

src/main/java/com/arangodb/springframework/transaction/ArangoTransactionManager.java

@@ -23,61 +23,85 @@
 import com.arangodb.ArangoDBException;
 import com.arangodb.ArangoDatabase;
 import com.arangodb.DbName;
-import com.arangodb.model.StreamTransactionOptions;
 import com.arangodb.springframework.core.ArangoOperations;
 import com.arangodb.springframework.repository.query.QueryTransactionBridge;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.lang.Nullable;
 import org.springframework.transaction.*;
 import org.springframework.transaction.support.AbstractPlatformTransactionManager;
 import org.springframework.transaction.support.DefaultTransactionStatus;
 import org.springframework.transaction.support.TransactionSynchronizationManager;
 
+import java.util.function.Function;
+
 /**
  * Transaction manager using ArangoDB stream transactions on the
- * {@linkplain ArangoOperations#getDatabaseName()} current database} of the template.
- * Isolation level {@linkplain TransactionDefinition#ISOLATION_SERIALIZABLE serializable} is not supported.
+ * {@linkplain ArangoOperations#getDatabaseName() current database} of the
+ * template. A {@linkplain ArangoTransactionObject transaction object} using
+ * a shared {@linkplain ArangoTransactionHolder holder} is used for the
+ * {@link DefaultTransactionStatus}. Neither
+ * {@linkplain TransactionDefinition#getPropagationBehavior() propagation}
+ * {@linkplain TransactionDefinition#PROPAGATION_NESTED nested} nor
+ * {@linkplain TransactionDefinition#getIsolationLevel() isolation}
+ * {@linkplain TransactionDefinition#ISOLATION_SERIALIZABLE serializable} are
+ * supported.
  */
-public class ArangoTransactionManager extends AbstractPlatformTransactionManager {
+public class ArangoTransactionManager extends AbstractPlatformTransactionManager implements InitializingBean {
 
     private final ArangoOperations operations;
     private final QueryTransactionBridge bridge;
 
     public ArangoTransactionManager(ArangoOperations operations, QueryTransactionBridge bridge) {
         this.operations = operations;
         this.bridge = bridge;
-        setValidateExistingTransaction(true);
+        super.setGlobalRollbackOnParticipationFailure(true);
+        super.setTransactionSynchronization(SYNCHRONIZATION_ON_ACTUAL_TRANSACTION);
+    }
+
+    /**
+     * Check for supported property settings.
+     */
+    @Override
+    public void afterPropertiesSet() {
+        if (isNestedTransactionAllowed()) {
+            throw new IllegalStateException("Nested transactions must not be allowed");
+        }
+        if (!isGlobalRollbackOnParticipationFailure()) {
+            throw new IllegalStateException("Global rollback on participating failure is needed");
+        }
+        if (getTransactionSynchronization() == SYNCHRONIZATION_NEVER) {
+            throw new IllegalStateException("Transaction synchronization is needed always");
+        }
     }
 
     /**
-     * Creates a new transaction object. Any synchronized resource will be reused.
+     * Creates a new transaction object. Any holder bound will be reused.
      */
     @Override
     protected ArangoTransactionObject doGetTransaction() {
         DbName database = operations.getDatabaseName();
-        if (logger.isDebugEnabled()) {
-            logger.debug("Create new transaction for database " + database);
-        }
+        ArangoTransactionHolder holder = (ArangoTransactionHolder) TransactionSynchronizationManager.getResource(database);
         try {
-            ArangoTransactionResource resource = (ArangoTransactionResource) TransactionSynchronizationManager.getResource(database);
-            return new ArangoTransactionObject(operations.driver().db(database), getDefaultTimeout(), resource);
+            return new ArangoTransactionObject(operations.driver().db(database), getDefaultTimeout(), holder);
         } catch (ArangoDBException error) {
             throw new TransactionSystemException("Cannot create transaction object", error);
         }
     }
 
     /**
-     * Configures the new transaction object. The resulting resource will be synchronized and the bridge will be initialized.
+     * Connect the new transaction object to the query bridge.
      *
-     * @see ArangoDatabase#beginStreamTransaction(StreamTransactionOptions)
-     * @see QueryTransactionBridge
+     * @see QueryTransactionBridge#setCurrentTransaction(Function)
+     * @see #prepareSynchronization(DefaultTransactionStatus, TransactionDefinition)
+     * @throws InvalidIsolationLevelException for {@link TransactionDefinition#ISOLATION_SERIALIZABLE}
      */
     @Override
-    protected void doBegin(Object transaction, TransactionDefinition definition) throws TransactionUsageException {
+    protected void doBegin(Object transaction, TransactionDefinition definition) throws InvalidIsolationLevelException {
         int isolationLevel = definition.getIsolationLevel();
-        if (isolationLevel != -1 && (isolationLevel & TransactionDefinition.ISOLATION_SERIALIZABLE) != 0) {
+        if (isolationLevel != TransactionDefinition.ISOLATION_DEFAULT && (isolationLevel & TransactionDefinition.ISOLATION_SERIALIZABLE) != 0) {
             throw new InvalidIsolationLevelException("ArangoDB does not support isolation level serializable");
         }
         ArangoTransactionObject tx = (ArangoTransactionObject) transaction;
-        tx.configure(definition);
         bridge.setCurrentTransaction(collections -> {
             try {
                 return tx.getOrBegin(collections).getStreamTransactionId();
@@ -88,9 +112,11 @@ protected void doBegin(Object transaction, TransactionDefinition definition) thr
     }
 
     /**
-     * Commit the current stream transaction iff any. The bridge is cleared afterwards.
+     * Commit the current stream transaction. The query bridge is cleared
+     * afterwards.
      *
      * @see ArangoDatabase#commitStreamTransaction(String)
+     * @see QueryTransactionBridge#clearCurrentTransaction()
      */
     @Override
     protected void doCommit(DefaultTransactionStatus status) throws TransactionException {
@@ -100,16 +126,19 @@ protected void doCommit(DefaultTransactionStatus status) throws TransactionExcep
         }
         try {
             tx.commit();
-            bridge.clearCurrentTransaction();
         } catch (ArangoDBException error) {
             throw new TransactionSystemException("Cannot commit transaction " + tx, error);
+        } finally {
+            bridge.clearCurrentTransaction();
         }
     }
 
     /**
-     * Roll back the current stream transaction iff any. The bridge is cleared afterwards.
+     * Roll back the current stream transaction. The query bridge is cleared
+     * afterwards.
      *
      * @see ArangoDatabase#abortStreamTransaction(String)
+     * @see QueryTransactionBridge#clearCurrentTransaction()
      */
     @Override
     protected void doRollback(DefaultTransactionStatus status) throws TransactionException {
@@ -119,58 +148,68 @@ protected void doRollback(DefaultTransactionStatus status) throws TransactionExc
         }
         try {
             tx.rollback();
-            bridge.clearCurrentTransaction();
         } catch (ArangoDBException error) {
             throw new TransactionSystemException("Cannot roll back transaction " + tx, error);
+        } finally {
+            bridge.clearCurrentTransaction();
         }
     }
 
     /**
-     * Check if the transaction objects has an underlying stream transaction.
-     *
-     * @see ArangoDatabase#getStreamTransaction(String)
+     * Check if the transaction object has the bound holder. For new
+     * transactions the holder will be bound afterwards.
      */
     @Override
     protected boolean isExistingTransaction(Object transaction) throws TransactionException {
-        return ((ArangoTransactionObject) transaction).exists();
+        ArangoTransactionHolder holder = ((ArangoTransactionObject) transaction).getHolder();
+        return holder == TransactionSynchronizationManager.getResource(operations.getDatabaseName());
     }
 
+    /**
+     * Mark the transaction as global rollback only.
+     *
+     * @see #isGlobalRollbackOnParticipationFailure()
+     */
     @Override
     protected void doSetRollbackOnly(DefaultTransactionStatus status) throws TransactionException {
         ArangoTransactionObject tx = (ArangoTransactionObject) status.getTransaction();
-        tx.setRollbackOnly();
+        tx.getHolder().setRollbackOnly();
     }
 
+    /**
+     * Any transaction object is configured according to the definition upfront.
+     *
+     * @see ArangoTransactionObject#configure(TransactionDefinition)
+     */
     @Override
-    protected DefaultTransactionStatus newTransactionStatus(TransactionDefinition definition, Object transaction, boolean newTransaction, boolean newSynchronization, boolean debug, Object suspendedResources) {
+    protected DefaultTransactionStatus newTransactionStatus(TransactionDefinition definition, @Nullable Object transaction, boolean newTransaction, boolean newSynchronization, boolean debug, @Nullable Object suspendedResources) {
+        if (transaction instanceof ArangoTransactionObject) {
+            ((ArangoTransactionObject) transaction).configure(definition);
+        }
         return super.newTransactionStatus(definition, transaction, newTransaction, newSynchronization, debug, suspendedResources);
     }
 
     /**
-     * Bind the resource for the first new transaction created.
+     * Bind the holder for the first new transaction created.
+     *
+     * @see ArangoTransactionHolder
      */
     @Override
     protected void prepareSynchronization(DefaultTransactionStatus status, TransactionDefinition definition) {
         super.prepareSynchronization(status, definition);
-        if (status.isNewTransaction()) {
-            ArangoTransactionResource resource = ((ArangoTransactionObject) status.getTransaction()).getResource();
-            resource.increaseReferences();
-            if (resource.isSingleReference()) {
-                TransactionSynchronizationManager.bindResource(operations.getDatabaseName(), resource);
-            }
+        if (status.isNewSynchronization()) {
+            ArangoTransactionHolder holder = ((ArangoTransactionObject) status.getTransaction()).getHolder();
+            TransactionSynchronizationManager.bindResource(operations.getDatabaseName(), holder);
         }
     }
 
     /**
-     * Unbind the resource for the last transaction completed.
+     * Unbind the holder from the last transaction completed.
+     *
+     * @see ArangoTransactionHolder
      */
     @Override
     protected void doCleanupAfterCompletion(Object transaction) {
-        ArangoTransactionResource resource = ((ArangoTransactionObject) transaction).getResource();
-        if (resource.isSingleReference()) {
-            TransactionSynchronizationManager.unbindResource(operations.getDatabaseName());
-        }
-        resource.decreasedReferences();
+        TransactionSynchronizationManager.unbindResource(operations.getDatabaseName());
     }
-
 }