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=29b52f75579b57e2fbad4563585de9b10828d154;hb=546cd1fd100dbaa36908b22c2f422320dbd8c4b2;hp=6d573dedf53223b34c8c2ca525bea3435d9bc812;hpb=f662ce8b1fa94b77ba66f7ece8bcaff91dee809e;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 6d573dedf5..29b52f7557 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;
@@ -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) {
@@ -215,12 +240,12 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
                 final DOMStoreReadTransaction readOnly = factory.newReadOnlyTransaction(parent.getIdentifier());
                 return new LocalTransactionContext(readOnly, parent.getIdentifier(), factory) {
                     @Override
-                    protected DOMStoreWriteTransaction getWriteDelegate() {
+                    DOMStoreWriteTransaction getWriteDelegate() {
                         throw new UnsupportedOperationException();
                     }
 
                     @Override
-                    protected DOMStoreReadTransaction getReadDelegate() {
+                    DOMStoreReadTransaction getReadDelegate() {
                         return readOnly;
                     }
                 };
@@ -228,12 +253,12 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
                 final DOMStoreReadWriteTransaction readWrite = factory.newReadWriteTransaction(parent.getIdentifier());
                 return new LocalTransactionContext(readWrite, parent.getIdentifier(), factory) {
                     @Override
-                    protected DOMStoreWriteTransaction getWriteDelegate() {
+                    DOMStoreWriteTransaction getWriteDelegate() {
                         return readWrite;
                     }
 
                     @Override
-                    protected DOMStoreReadTransaction getReadDelegate() {
+                    DOMStoreReadTransaction getReadDelegate() {
                         return readWrite;
                     }
                 };
@@ -241,12 +266,12 @@ abstract class AbstractTransactionContextFactory<F extends LocalTransactionFacto
                 final DOMStoreWriteTransaction writeOnly = factory.newWriteOnlyTransaction(parent.getIdentifier());
                 return new LocalTransactionContext(writeOnly, parent.getIdentifier(), factory) {
                     @Override
-                    protected DOMStoreWriteTransaction getWriteDelegate() {
+                    DOMStoreWriteTransaction getWriteDelegate() {
                         return writeOnly;
                     }
 
                     @Override
-                    protected DOMStoreReadTransaction getReadDelegate() {
+                    DOMStoreReadTransaction getReadDelegate() {
                         throw new UnsupportedOperationException();
                     }
                 };