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 java.util.ArrayList;
import java.util.Iterator;
private static final MessageSupplier COMMIT_MESSAGE_SUPPLIER = new MessageSupplier() {
@Override
- public Object newMessage(TransactionIdentifier transactionId, short version) {
+ public Object newMessage(final TransactionIdentifier transactionId, final short version) {
return new CommitTransaction(transactionId, version).toSerializable();
}
@Override
- public boolean isSerializedReplyType(Object reply) {
+ public boolean isSerializedReplyType(final Object reply) {
return CommitTransactionReply.isSerializedType(reply);
}
};
private static final MessageSupplier ABORT_MESSAGE_SUPPLIER = new MessageSupplier() {
@Override
- public Object newMessage(TransactionIdentifier transactionId, short version) {
+ public Object newMessage(final TransactionIdentifier transactionId, final short version) {
return new AbortTransaction(transactionId, version).toSerializable();
}
@Override
- public boolean isSerializedReplyType(Object reply) {
+ public boolean isSerializedReplyType(final Object reply) {
return AbortTransactionReply.isSerializedType(reply);
}
};
private final TransactionIdentifier transactionId;
private volatile OperationCallback commitOperationCallback;
- public ThreePhaseCommitCohortProxy(ActorContext actorContext, List<CohortInfo> cohorts,
- TransactionIdentifier transactionId) {
+ public ThreePhaseCommitCohortProxy(final ActorContext actorContext, final List<CohortInfo> cohorts,
+ final TransactionIdentifier transactionId) {
this.actorContext = actorContext;
this.cohorts = cohorts;
this.transactionId = Preconditions.checkNotNull(transactionId);
}
final AtomicInteger completed = new AtomicInteger(cohorts.size());
+ final Object lock = new Object();
for (final CohortInfo info: cohorts) {
info.getActorFuture().onComplete(new OnComplete<ActorSelection>() {
@Override
- public void onComplete(Throwable failure, ActorSelection actor) {
- synchronized (completed) {
+ public void onComplete(final Throwable failure, final ActorSelection actor) {
+ synchronized (lock) {
boolean done = completed.decrementAndGet() == 0;
if (failure != null) {
LOG.debug("Tx {}: a cohort Future failed", transactionId, failure);
Futures.addCallback(resolveCohorts(), new FutureCallback<Void>() {
@Override
- public void onSuccess(Void notUsed) {
+ public void onSuccess(final Void notUsed) {
finishCanCommit(returnFuture);
}
@Override
- public void onFailure(Throwable failure) {
+ public void onFailure(final Throwable failure) {
returnFuture.setException(failure);
}
- });
+ }, MoreExecutors.directExecutor());
return returnFuture;
}
final OnComplete<Object> onComplete = new OnComplete<Object>() {
@Override
- public void onComplete(Throwable failure, Object response) {
+ public void onComplete(final Throwable failure, final Object response) {
if (failure != null) {
LOG.debug("Tx {}: a canCommit cohort Future failed", transactionId, failure);
sendCanCommitTransaction(iterator.next(), onComplete);
}
- private void sendCanCommitTransaction(CohortInfo toCohortInfo, OnComplete<Object> onComplete) {
+ private void sendCanCommitTransaction(final CohortInfo toCohortInfo, final OnComplete<Object> onComplete) {
CanCommitTransaction message = new CanCommitTransaction(transactionId, toCohortInfo.getActorVersion());
LOG.debug("Tx {}: sending {} to {}", transactionId, message, toCohortInfo.getResolvedActor());
future.onComplete(onComplete, actorContext.getClientDispatcher());
}
- private Future<Iterable<Object>> invokeCohorts(MessageSupplier messageSupplier) {
+ private Future<Iterable<Object>> invokeCohorts(final MessageSupplier messageSupplier) {
List<Future<Object>> futureList = Lists.newArrayListWithCapacity(cohorts.size());
for (CohortInfo cohort : cohorts) {
Object message = messageSupplier.newMessage(transactionId, cohort.getActorVersion());
- LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message , cohort);
+ LOG.debug("Tx {}: Sending {} to cohort {}", transactionId, message , cohort.getResolvedActor());
futureList.add(actorContext.executeOperationAsync(cohort.getResolvedActor(), message,
actorContext.getTransactionCommitOperationTimeout()));
}
@SuppressWarnings("checkstyle:IllegalCatch")
- private static boolean successfulFuture(ListenableFuture<Void> future) {
+ private static boolean successfulFuture(final ListenableFuture<Void> future) {
if (!future.isDone()) {
return false;
}
} else {
Futures.addCallback(future, new FutureCallback<Void>() {
@Override
- public void onSuccess(Void notUsed) {
+ public void onSuccess(final Void notUsed) {
finishVoidOperation(operationName, messageSupplier, expectedResponseClass,
propagateException, returnFuture, callback);
}
@Override
- public void onFailure(Throwable failure) {
- LOG.debug("Tx {}: a {} cohort path Future failed: {}", transactionId, operationName, failure);
+ public void onFailure(final Throwable failure) {
+ LOG.debug("Tx {}: a {} cohort path Future failed", transactionId, operationName, failure);
if (propagateException) {
returnFuture.setException(failure);
returnFuture.set(null);
}
}
- });
+ }, MoreExecutors.directExecutor());
}
return returnFuture;
}
- private void finishVoidOperation(final String operationName, MessageSupplier messageSupplier,
+ private void finishVoidOperation(final String operationName, final MessageSupplier messageSupplier,
final Class<?> expectedResponseClass, final boolean propagateException,
final SettableFuture<Void> returnFuture, final OperationCallback callback) {
LOG.debug("Tx {} finish {}", transactionId, operationName);
combinedFuture.onComplete(new OnComplete<Iterable<Object>>() {
@Override
- public void onComplete(Throwable failure, 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) {
private volatile ActorSelection resolvedActor;
private final Supplier<Short> actorVersionSupplier;
- CohortInfo(Future<ActorSelection> actorFuture, Supplier<Short> actorVersionSupplier) {
+ CohortInfo(final Future<ActorSelection> actorFuture, final Supplier<Short> actorVersionSupplier) {
this.actorFuture = actorFuture;
this.actorVersionSupplier = actorVersionSupplier;
}
return resolvedActor;
}
- void setResolvedActor(ActorSelection resolvedActor) {
+ void setResolvedActor(final ActorSelection resolvedActor) {
this.resolvedActor = resolvedActor;
}