Specialize TransactionContextWrapper

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / AbstractTransactionContextFactory.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java
index 6d573dedf53223b34c8c2ca525bea3435d9bc812..211ebec63c3074e9199d9c034290928495451b91 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractTransactionContextFactory.java
@@ -10,7 +10,6 @@ package org.opendaylight.controller.cluster.datastore;
 import static java.util.Objects.requireNonNull;
 
 import akka.actor.ActorSelection;
-import akka.dispatch.OnComplete;
 import java.util.Collection;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
@@ -80,20 +79,48 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
         return null;
     }
 
-    private void onFindPrimaryShardSuccess(PrimaryShardInfo primaryShardInfo, TransactionProxy parent,
-            String shardName, TransactionContextWrapper transactionContextWrapper) {
+    private AbstractTransactionContextWrapper maybeCreateDirectTransactionContextWrapper(
+            final PrimaryShardInfo primaryShardInfo, final TransactionProxy parent,
+            final String shardName, final DelayedTransactionContextWrapper transactionContextWrapper) {
+        LOG.debug("Tx {}: Found primary {} for shard {}, trying to use DirectTransactionContextWrapper",
+                parent.getIdentifier(), primaryShardInfo.getPrimaryShardActor(), shardName);
+
+        updateShardInfo(shardName, primaryShardInfo);
+
+        final TransactionContext localContext = maybeCreateLocalTransactionContext(parent, shardName);
+        try {
+            if (localContext != null) {
+                LOG.debug("Tx {}: Local transaction context created successfully, using DirectTransactionWrapper",
+                        parent.getIdentifier());
+                return new DirectTransactionContextWrapper(parent.getIdentifier(), actorUtils, shardName,
+                        localContext);
+            } else {
+                LOG.debug("Tx {}: Local transaction context creation failed, using DelayedTransactionWrapper",
+                        parent.getIdentifier());
+                final RemoteTransactionContextSupport remote = new RemoteTransactionContextSupport(
+                        transactionContextWrapper, parent, shardName);
+                remote.setPrimaryShard(primaryShardInfo);
+                return transactionContextWrapper;
+            }
+        } finally {
+            onTransactionContextCreated(parent.getIdentifier());
+        }
+    }
+
+    private void onFindPrimaryShardSuccess(final PrimaryShardInfo primaryShardInfo, final TransactionProxy parent,
+            final String shardName, final DelayedTransactionContextWrapper transactionContextWrapper) {
         LOG.debug("Tx {}: Found primary {} for shard {}", parent.getIdentifier(),
                 primaryShardInfo.getPrimaryShardActor(), shardName);
 
         updateShardInfo(shardName, primaryShardInfo);
 
+        final TransactionContext localContext = maybeCreateLocalTransactionContext(parent, shardName);
         try {
-            TransactionContext localContext = maybeCreateLocalTransactionContext(parent, shardName);
             if (localContext != null) {
                 transactionContextWrapper.executePriorTransactionOperations(localContext);
             } else {
-                RemoteTransactionContextSupport remote = new RemoteTransactionContextSupport(transactionContextWrapper,
-                        parent, shardName);
+                final RemoteTransactionContextSupport remote = new RemoteTransactionContextSupport(
+                        transactionContextWrapper, parent, shardName);
                 remote.setPrimaryShard(primaryShardInfo);
             }
         } finally {
@@ -101,45 +128,43 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
         }
     }
 
-    private void onFindPrimaryShardFailure(Throwable failure, TransactionProxy parent,
-            String shardName, TransactionContextWrapper transactionContextWrapper) {
+    private void onFindPrimaryShardFailure(final Throwable failure, final TransactionProxy parent,
+            final String shardName, final DelayedTransactionContextWrapper transactionContextWrapper) {
         LOG.debug("Tx {}: Find primary for shard {} failed", parent.getIdentifier(), shardName, failure);
 
         try {
-            transactionContextWrapper.executePriorTransactionOperations(new NoOpTransactionContext(failure,
-                    parent.getIdentifier()));
+            transactionContextWrapper.executePriorTransactionOperations(
+                    new NoOpTransactionContext(failure, parent.getIdentifier()));
         } finally {
             onTransactionContextCreated(parent.getIdentifier());
         }
     }
 
-    final TransactionContextWrapper newTransactionContextWrapper(final TransactionProxy parent,
+    final AbstractTransactionContextWrapper newTransactionContextWrapper(final TransactionProxy parent,
             final String shardName) {
-        final TransactionContextWrapper transactionContextWrapper =
-                new TransactionContextWrapper(parent.getIdentifier(), actorUtils, shardName);
-
-        Future<PrimaryShardInfo> findPrimaryFuture = findPrimaryShard(shardName, parent.getIdentifier());
+        final DelayedTransactionContextWrapper contextWrapper = new DelayedTransactionContextWrapper(
+                parent.getIdentifier(), actorUtils, shardName);
+        final Future<PrimaryShardInfo> findPrimaryFuture = findPrimaryShard(shardName, parent.getIdentifier());
         if (findPrimaryFuture.isCompleted()) {
-            Try<PrimaryShardInfo> maybe = findPrimaryFuture.value().get();
+            final Try<PrimaryShardInfo> maybe = findPrimaryFuture.value().get();
             if (maybe.isSuccess()) {
-                onFindPrimaryShardSuccess(maybe.get(), parent, shardName, transactionContextWrapper);
+                return maybeCreateDirectTransactionContextWrapper(maybe.get(), parent, shardName,
+                        contextWrapper);
             } else {
-                onFindPrimaryShardFailure(maybe.failed().get(), parent, shardName, transactionContextWrapper);
+                onFindPrimaryShardFailure(maybe.failed().get(), parent, shardName,
+                        contextWrapper);
             }
         } else {
-            findPrimaryFuture.onComplete(new OnComplete<PrimaryShardInfo>() {
-                @Override
-                public void onComplete(final Throwable failure, final PrimaryShardInfo primaryShardInfo) {
-                    if (failure == null) {
-                        onFindPrimaryShardSuccess(primaryShardInfo, parent, shardName, transactionContextWrapper);
-                    } else {
-                        onFindPrimaryShardFailure(failure, parent, shardName, transactionContextWrapper);
-                    }
+            findPrimaryFuture.onComplete((result) -> {
+                if (result.isSuccess()) {
+                    onFindPrimaryShardSuccess(result.get(), parent, shardName, contextWrapper);
+                } else {
+                    onFindPrimaryShardFailure(result.failed().get(), parent, shardName, contextWrapper);
                 }
+                return null;
             }, actorUtils.getClientDispatcher());
         }
-
-        return transactionContextWrapper;
+        return contextWrapper;
     }
 
     private void updateShardInfo(final String shardName, final PrimaryShardInfo primaryShardInfo) {