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%2FThreePhaseCommitCohortProxy.java;h=9505894b4b1cad73fd0d264ee8b7ba641f26ae66;hb=1e8d188e98614a1f3d781b2f80d61fcd0afde368;hp=4d80d7fd8ab3fb3c5200f62aa4c589a6343ad2b7;hpb=634dfac8eead60f443bf75e749c70d1f2bb29198;p=controller.git 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 4d80d7fd8a..9505894b4b 100644 --- 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,12 +5,13 @@ * 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; @@ -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 cohorts; private final SettableFuture cohortsResolvedFuture = SettableFuture.create(); private final TransactionIdentifier transactionId; private volatile OperationCallback commitOperationCallback; - public ThreePhaseCommitCohortProxy(final ActorContext actorContext, final List cohorts, + public ThreePhaseCommitCohortProxy(final ActorUtils actorUtils, final List 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; @@ -152,7 +153,7 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< return; } - commitOperationCallback = new TransactionRateLimitingCallback(actorContext); + commitOperationCallback = new TransactionRateLimitingCallback(actorUtils); commitOperationCallback.run(); final Iterator iterator = cohorts.iterator(); @@ -206,9 +207,9 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< LOG.debug("Tx {}: sending {} to {}", transactionId, message, toCohortInfo.getResolvedActor()); - Future future = actorContext.executeOperationAsync(toCohortInfo.getResolvedActor(), - message.toSerializable(), actorContext.getTransactionCommitOperationTimeout()); - future.onComplete(onComplete, actorContext.getClientDispatcher()); + Future future = actorUtils.executeOperationAsync(toCohortInfo.getResolvedActor(), + message.toSerializable(), actorUtils.getTransactionCommitOperationTimeout()); + future.onComplete(onComplete, actorUtils.getClientDispatcher()); } private Future> invokeCohorts(final MessageSupplier messageSupplier) { @@ -218,11 +219,11 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< 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 @@ -350,7 +351,7 @@ public class ThreePhaseCommitCohortProxy extends AbstractThreePhaseCommitCohort< callback.success(); } } - }, actorContext.getClientDispatcher()); + }, actorUtils.getClientDispatcher()); } @Override @@ -386,8 +387,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(); } }