X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FAbstractTransactionContextFactory.java;h=211ebec63c3074e9199d9c034290928495451b91;hb=refs%2Fchanges%2F64%2F90864%2F56;hp=a415a8c087fc6f28e5bff6c8b0406bb03b3c6117;hpb=127042ea7e148d9dc0282acc3780b4754ca69e12;p=controller.git

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 a415a8c087..211ebec63c 100644
--- 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;
@@ -25,7 +24,7 @@ import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
 import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTree;
+import org.opendaylight.yangtools.yang.data.api.schema.tree.ReadOnlyDataTree;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.concurrent.Future;
@@ -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,49 +128,47 @@ 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) {
-        final Optional<DataTree> maybeDataTree = primaryShardInfo.getLocalShardDataTree();
+        final Optional<ReadOnlyDataTree> maybeDataTree = primaryShardInfo.getLocalShardDataTree();
         if (maybeDataTree.isPresent()) {
             if (!knownLocal.containsKey(shardName)) {
                 LOG.debug("Shard {} resolved to local data tree - adding local factory", shardName);
@@ -190,7 +215,7 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
      *                 read-only manner.
      * @return Transaction factory for local use.
      */
-    protected abstract F factoryForShard(String shardName, ActorSelection shardLeader, DataTree dataTree);
+    protected abstract F factoryForShard(String shardName, ActorSelection shardLeader, ReadOnlyDataTree dataTree);
 
     /**
      * Callback invoked from child transactions to push any futures, which need to