* 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;
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;
}
};
- 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);
}
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
}
return cohortsResolvedFuture;
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);
return;
}
- commitOperationCallback = new TransactionRateLimitingCallback(actorContext);
+ commitOperationCallback = new TransactionRateLimitingCallback(actorUtils);
commitOperationCallback.run();
final Iterator<CohortInfo> iterator = cohorts.iterator();
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
@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);
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) {
callback.success();
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
}
@Override
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;
}
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();
}
}