Migrate DatastoreContextIntrospectorFactory to OSGi DS

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

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxy.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxy.java
index 90dcec238aa13f302d429aa1b2e38c7d78b81d67..5f82584f856cc69b987c473178232d3aea6b3a66 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxy.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohortProxy.java
@@ -5,23 +5,24 @@
  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
  * and is available at http://www.eclipse.org/legal/epl-v10.html
  */
-
 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 akka.dispatch.OnComplete;
-import com.google.common.base.Preconditions;
-import com.google.common.base.Supplier;
-import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
@@ -29,7 +30,7 @@ import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransacti
 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
 import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
-import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.concurrent.Future;
@@ -65,17 +66,17 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<
         }
     };
 
-    private final ActorContext actorContext;
+    private final ActorUtils actorUtils;
     private final List<CohortInfo> cohorts;
     private final SettableFuture<Void> cohortsResolvedFuture = SettableFuture.create();
     private final TransactionIdentifier transactionId;
     private volatile OperationCallback commitOperationCallback;
 
-    public ThreePhaseCommitCohortProxy(final ActorContext actorContext, final List<CohortInfo> cohorts,
+    public ThreePhaseCommitCohortProxy(final ActorUtils actorUtils, final List<CohortInfo> cohorts,
             final TransactionIdentifier transactionId) {
-        this.actorContext = actorContext;
+        this.actorUtils = actorUtils;
         this.cohorts = cohorts;
-        this.transactionId = Preconditions.checkNotNull(transactionId);
+        this.transactionId = requireNonNull(transactionId);
 
         if (cohorts.isEmpty()) {
             cohortsResolvedFuture.set(null);
@@ -109,7 +110,7 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<
                         }
                     }
                 }
-            }, actorContext.getClientDispatcher());
+            }, actorUtils.getClientDispatcher());
         }
 
         return cohortsResolvedFuture;
@@ -142,6 +143,8 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<
         return returnFuture;
     }
 
+    @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+            justification = "https://github.com/spotbugs/spotbugs/issues/811")
     private void finishCanCommit(final SettableFuture<Boolean> returnFuture) {
         LOG.debug("Tx {} finishCanCommit", transactionId);
 
@@ -152,7 +155,7 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<
             return;
         }
 
-        commitOperationCallback = new TransactionRateLimitingCallback(actorContext);
+        commitOperationCallback = new TransactionRateLimitingCallback(actorUtils);
         commitOperationCallback.run();
 
         final Iterator<CohortInfo> iterator = cohorts.iterator();
@@ -206,23 +209,23 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<
 
         LOG.debug("Tx {}: sending {} to {}", transactionId, message, toCohortInfo.getResolvedActor());
 
-        Future<Object> future = actorContext.executeOperationAsync(toCohortInfo.getResolvedActor(),
-                message.toSerializable(), actorContext.getTransactionCommitOperationTimeout());
-        future.onComplete(onComplete, actorContext.getClientDispatcher());
+        Future<Object> future = actorUtils.executeOperationAsync(toCohortInfo.getResolvedActor(),
+                message.toSerializable(), actorUtils.getTransactionCommitOperationTimeout());
+        future.onComplete(onComplete, actorUtils.getClientDispatcher());
     }
 
     private Future<Iterable<Object>> invokeCohorts(final MessageSupplier messageSupplier) {
-        List<Future<Object>> futureList = Lists.newArrayListWithCapacity(cohorts.size());
+        List<Future<Object>> futureList = new ArrayList<>(cohorts.size());
         for (CohortInfo cohort : cohorts) {
             Object message = messageSupplier.newMessage(transactionId, cohort.getActorVersion());
 
             LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message , cohort.getResolvedActor());
 
-            futureList.add(actorContext.executeOperationAsync(cohort.getResolvedActor(), message,
-                    actorContext.getTransactionCommitOperationTimeout()));
+            futureList.add(actorUtils.executeOperationAsync(cohort.getResolvedActor(), message,
+                    actorUtils.getTransactionCommitOperationTimeout()));
         }
 
-        return akka.dispatch.Futures.sequence(futureList, actorContext.getClientDispatcher());
+        return akka.dispatch.Futures.sequence(futureList, actorUtils.getClientDispatcher());
     }
 
     @Override
@@ -291,7 +294,7 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<
 
                 @Override
                 public void onFailure(final Throwable failure) {
-                    LOG.debug("Tx {}: a {} cohort path Future failed: {}", transactionId, operationName, failure);
+                    LOG.debug("Tx {}: a {} cohort path Future failed", transactionId, operationName, failure);
 
                     if (propagateException) {
                         returnFuture.setException(failure);
@@ -316,7 +319,7 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<
 
         combinedFuture.onComplete(new OnComplete<Iterable<Object>>() {
             @Override
-            public void onComplete(final Throwable failure, final Iterable<Object> responses) throws Throwable {
+            public void onComplete(final Throwable failure, final Iterable<Object> responses) {
                 Throwable exceptionToPropagate = failure;
                 if (exceptionToPropagate == null) {
                     for (Object response: responses) {
@@ -350,7 +353,7 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<
                     callback.success();
                 }
             }
-        }, actorContext.getClientDispatcher());
+        }, actorUtils.getClientDispatcher());
     }
 
     @Override
@@ -365,9 +368,10 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<
 
     static class CohortInfo {
         private final Future<ActorSelection> actorFuture;
-        private volatile ActorSelection resolvedActor;
         private final Supplier<Short> actorVersionSupplier;
 
+        private volatile ActorSelection resolvedActor;
+
         CohortInfo(final Future<ActorSelection> actorFuture, final Supplier<Short> actorVersionSupplier) {
             this.actorFuture = actorFuture;
             this.actorVersionSupplier = actorVersionSupplier;
@@ -386,8 +390,7 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort<
         }
 
         short getActorVersion() {
-            Preconditions.checkState(resolvedActor != null,
-                    "getActorVersion cannot be called until the actor is resolved");
+            checkState(resolvedActor != null, "getActorVersion cannot be called until the actor is resolved");
             return actorVersionSupplier.get();
         }
     }