X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FTransactionProxy.java;h=019b0dbebe20c18f964b6f8c5f0f385bf59d3a76;hp=b04dd29a5888ff716fdcd3ba8ad038800a36613e;hb=66e553f2098ea61426c4a441be57395673535c2f;hpb=731e7284cf0895fdb1b89427f91762e80e67c2ff

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java
index b04dd29a58..019b0dbebe 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/TransactionProxy.java
@@ -7,14 +7,13 @@
  */
 package org.opendaylight.controller.cluster.datastore;
 
+import static com.google.common.base.Preconditions.checkState;
+import static java.util.Objects.requireNonNull;
+
 import akka.actor.ActorSelection;
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Function;
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Supplier;
 import com.google.common.collect.Iterables;
-import com.google.common.util.concurrent.CheckedFuture;
+import com.google.common.util.concurrent.FluentFuture;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
@@ -24,6 +23,7 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Optional;
 import java.util.Set;
 import java.util.SortedSet;
 import java.util.TreeMap;
@@ -36,10 +36,8 @@ import org.opendaylight.controller.cluster.datastore.modification.AbstractModifi
 import org.opendaylight.controller.cluster.datastore.modification.DeleteModification;
 import org.opendaylight.controller.cluster.datastore.modification.MergeModification;
 import org.opendaylight.controller.cluster.datastore.modification.WriteModification;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
 import org.opendaylight.controller.cluster.datastore.utils.NormalizedNodeAggregator;
-import org.opendaylight.mdsal.common.api.MappingCheckedFuture;
-import org.opendaylight.mdsal.common.api.ReadFailedException;
 import org.opendaylight.mdsal.dom.spi.store.AbstractDOMStoreTransaction;
 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
@@ -70,23 +68,21 @@ public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIde
 
     @VisibleForTesting
     public TransactionProxy(final AbstractTransactionContextFactory<?> txContextFactory, final TransactionType type) {
-        super(txContextFactory.nextIdentifier(), txContextFactory.getActorContext().getDatastoreContext()
+        super(txContextFactory.nextIdentifier(), txContextFactory.getActorUtils().getDatastoreContext()
                 .isTransactionDebugContextEnabled());
         this.txContextFactory = txContextFactory;
-        this.type = Preconditions.checkNotNull(type);
+        this.type = requireNonNull(type);
 
         LOG.debug("New {} Tx - {}", type, getIdentifier());
     }
 
     @Override
-    public CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
+    public FluentFuture<Boolean> exists(final YangInstanceIdentifier path) {
         return executeRead(shardNameFromIdentifier(path), new DataExists(path, DataStoreVersions.CURRENT_VERSION));
     }
 
-    private <T> CheckedFuture<T, ReadFailedException> executeRead(final String shardName,
-            final AbstractRead<T> readCmd) {
-        Preconditions.checkState(type != TransactionType.WRITE_ONLY,
-                "Reads from write-only transactions are not allowed");
+    private <T> FluentFuture<T> executeRead(final String shardName, final AbstractRead<T> readCmd) {
+        checkState(type != TransactionType.WRITE_ONLY, "Reads from write-only transactions are not allowed");
 
         LOG.trace("Tx {} {} {}", getIdentifier(), readCmd.getClass().getSimpleName(), readCmd.getPath());
 
@@ -99,48 +95,45 @@ public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIde
             }
         });
 
-        return MappingCheckedFuture.create(proxyFuture, ReadFailedException.MAPPER);
+        return FluentFuture.from(proxyFuture);
     }
 
     @Override
-    public CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
-        Preconditions.checkState(type != TransactionType.WRITE_ONLY,
-                "Reads from write-only transactions are not allowed");
-        Preconditions.checkNotNull(path, "path should not be null");
+    public FluentFuture<Optional<NormalizedNode<?, ?>>> read(final YangInstanceIdentifier path) {
+        checkState(type != TransactionType.WRITE_ONLY, "Reads from write-only transactions are not allowed");
+        requireNonNull(path, "path should not be null");
 
         LOG.trace("Tx {} read {}", getIdentifier(), path);
         return path.isEmpty() ? readAllData() :  singleShardRead(shardNameFromIdentifier(path), path);
     }
 
-    private CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> singleShardRead(
+    private FluentFuture<Optional<NormalizedNode<?, ?>>> singleShardRead(
             final String shardName, final YangInstanceIdentifier path) {
         return executeRead(shardName, new ReadData(path, DataStoreVersions.CURRENT_VERSION));
     }
 
-    private CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> readAllData() {
-        final Set<String> allShardNames = txContextFactory.getActorContext().getConfiguration().getAllShardNames();
-        final Collection<CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException>> futures =
-                new ArrayList<>(allShardNames.size());
+    private FluentFuture<Optional<NormalizedNode<?, ?>>> readAllData() {
+        final Set<String> allShardNames = txContextFactory.getActorUtils().getConfiguration().getAllShardNames();
+        final Collection<FluentFuture<Optional<NormalizedNode<?, ?>>>> futures = new ArrayList<>(allShardNames.size());
 
         for (String shardName : allShardNames) {
-            futures.add(singleShardRead(shardName, YangInstanceIdentifier.EMPTY));
+            futures.add(singleShardRead(shardName, YangInstanceIdentifier.empty()));
         }
 
         final ListenableFuture<List<Optional<NormalizedNode<?, ?>>>> listFuture = Futures.allAsList(futures);
         final ListenableFuture<Optional<NormalizedNode<?, ?>>> aggregateFuture;
 
-        aggregateFuture = Futures.transform(listFuture,
-            (Function<List<Optional<NormalizedNode<?, ?>>>, Optional<NormalizedNode<?, ?>>>) input -> {
-                try {
-                    return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.EMPTY, input,
-                            txContextFactory.getActorContext().getSchemaContext(),
-                            txContextFactory.getActorContext().getDatastoreContext().getLogicalStoreType());
-                } catch (DataValidationFailedException e) {
-                    throw new IllegalArgumentException("Failed to aggregate", e);
-                }
-            }, MoreExecutors.directExecutor());
+        aggregateFuture = Futures.transform(listFuture, input -> {
+            try {
+                return NormalizedNodeAggregator.aggregate(YangInstanceIdentifier.empty(), input,
+                        txContextFactory.getActorUtils().getSchemaContext(),
+                        txContextFactory.getActorUtils().getDatastoreContext().getLogicalStoreType());
+            } catch (DataValidationFailedException e) {
+                throw new IllegalArgumentException("Failed to aggregate", e);
+            }
+        }, MoreExecutors.directExecutor());
 
-        return MappingCheckedFuture.create(aggregateFuture, ReadFailedException.MAPPER);
+        return FluentFuture.from(aggregateFuture);
     }
 
     @Override
@@ -174,10 +167,8 @@ public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIde
     }
 
     private void checkModificationState() {
-        Preconditions.checkState(type != TransactionType.READ_ONLY,
-                "Modification operation on read-only transaction is not allowed");
-        Preconditions.checkState(state == TransactionState.OPEN,
-                "Transaction is sealed - further modifications are not allowed");
+        checkState(type != TransactionType.READ_ONLY, "Modification operation on read-only transaction is not allowed");
+        checkState(state == TransactionState.OPEN, "Transaction is sealed - further modifications are not allowed");
     }
 
     private boolean seal(final TransactionState newState) {
@@ -192,7 +183,7 @@ public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIde
     @Override
     public final void close() {
         if (!seal(TransactionState.CLOSED)) {
-            Preconditions.checkState(state == TransactionState.CLOSED, "Transaction %s is ready, it cannot be closed",
+            checkState(state == TransactionState.CLOSED, "Transaction %s is ready, it cannot be closed",
                 getIdentifier());
             // Idempotent no-op as per AutoCloseable recommendation
             return;
@@ -213,10 +204,10 @@ public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIde
 
     @Override
     public final AbstractThreePhaseCommitCohort<?> ready() {
-        Preconditions.checkState(type != TransactionType.READ_ONLY, "Read-only transactions cannot be readied");
+        checkState(type != TransactionType.READ_ONLY, "Read-only transactions cannot be readied");
 
         final boolean success = seal(TransactionState.READY);
-        Preconditions.checkState(success, "Transaction %s is %s, it cannot be readied", getIdentifier(), state);
+        checkState(success, "Transaction %s is %s, it cannot be readied", getIdentifier(), state);
 
         LOG.debug("Tx {} Readying {} components for commit", getIdentifier(), txContextWrappers.size());
 
@@ -266,14 +257,14 @@ public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIde
             future = getDirectCommitFuture(transactionContext, operationCallbackRef, null);
         }
 
-        return new SingleCommitCohortProxy(txContextFactory.getActorContext(), future, getIdentifier(),
+        return new SingleCommitCohortProxy(txContextFactory.getActorUtils(), future, getIdentifier(),
             operationCallbackRef);
     }
 
     private Future<?> getDirectCommitFuture(final TransactionContext transactionContext,
             final OperationCallback.Reference operationCallbackRef, final Boolean havePermit) {
         TransactionRateLimitingCallback rateLimitingCallback = new TransactionRateLimitingCallback(
-                txContextFactory.getActorContext());
+                txContextFactory.getActorUtils());
         operationCallbackRef.set(rateLimitingCallback);
         rateLimitingCallback.run();
         return transactionContext.directCommit(havePermit);
@@ -282,8 +273,7 @@ public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIde
     private AbstractThreePhaseCommitCohort<ActorSelection> createMultiCommitCohort() {
 
         final List<ThreePhaseCommitCohortProxy.CohortInfo> cohorts = new ArrayList<>(txContextWrappers.size());
-        final java.util.Optional<SortedSet<String>> shardNames =
-                java.util.Optional.of(new TreeSet<>(txContextWrappers.keySet()));
+        final Optional<SortedSet<String>> shardNames = Optional.of(new TreeSet<>(txContextWrappers.keySet()));
         for (Entry<String, TransactionContextWrapper> e : txContextWrappers.entrySet()) {
             LOG.debug("Tx {} Readying transaction for shard {}", getIdentifier(), e.getKey());
 
@@ -292,17 +282,15 @@ public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIde
             // The remote tx version is obtained the via TransactionContext which may not be available yet so
             // we pass a Supplier to dynamically obtain it. Once the ready Future is resolved the
             // TransactionContext is available.
-            Supplier<Short> txVersionSupplier = () -> wrapper.getTransactionContext().getTransactionVersion();
-
             cohorts.add(new ThreePhaseCommitCohortProxy.CohortInfo(wrapper.readyTransaction(shardNames),
-                    txVersionSupplier));
+                () -> wrapper.getTransactionContext().getTransactionVersion()));
         }
 
-        return new ThreePhaseCommitCohortProxy(txContextFactory.getActorContext(), cohorts, getIdentifier());
+        return new ThreePhaseCommitCohortProxy(txContextFactory.getActorUtils(), cohorts, getIdentifier());
     }
 
     private String shardNameFromIdentifier(final YangInstanceIdentifier path) {
-        return txContextFactory.getActorContext().getShardStrategyFactory().getStrategy(path).findShard(path);
+        return txContextFactory.getActorUtils().getShardStrategyFactory().getStrategy(path).findShard(path);
     }
 
     private TransactionContextWrapper getContextWrapper(final YangInstanceIdentifier path) {
@@ -328,7 +316,7 @@ public class TransactionProxy extends AbstractDOMStoreTransaction<TransactionIde
         return state != TransactionState.OPEN;
     }
 
-    ActorContext getActorContext() {
-        return txContextFactory.getActorContext();
+    ActorUtils getActorUtils() {
+        return txContextFactory.getActorUtils();
     }
 }