import akka.dispatch.OnComplete;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ListenableFuture;
-import org.opendaylight.controller.cluster.datastore.identifiers.TransactionIdentifier;
+import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
import org.opendaylight.controller.cluster.datastore.messages.CommitTransactionReply;
import org.opendaylight.controller.cluster.datastore.messages.ReadyLocalTransaction;
import org.opendaylight.controller.cluster.datastore.utils.ActorContext;
private final DataTreeModification modification;
private final ActorContext actorContext;
private final ActorSelection leader;
- private Exception operationError;
+ private final Exception operationError;
protected LocalThreePhaseCommitCohort(final ActorContext actorContext, final ActorSelection leader,
- final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction, final DataTreeModification modification) {
+ final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction,
+ final DataTreeModification modification) {
this.actorContext = Preconditions.checkNotNull(actorContext);
this.leader = Preconditions.checkNotNull(leader);
this.transaction = Preconditions.checkNotNull(transaction);
this.modification = Preconditions.checkNotNull(modification);
+ this.operationError = null;
}
protected LocalThreePhaseCommitCohort(final ActorContext actorContext, final ActorSelection leader,
}
private Future<Object> initiateCommit(final boolean immediate) {
- if(operationError != null) {
+ if (operationError != null) {
return Futures.failed(operationError);
}
- final ReadyLocalTransaction message = new ReadyLocalTransaction(transaction.getIdentifier().toString(),
+ final ReadyLocalTransaction message = new ReadyLocalTransaction(transaction.getIdentifier(),
modification, immediate);
return actorContext.executeOperationAsync(leader, message, actorContext.getTransactionCommitOperationTimeout());
}
- void setOperationError(Exception operationError) {
- this.operationError = operationError;
- }
-
Future<ActorSelection> initiateCoordinatedCommit() {
final Future<Object> messageFuture = initiateCommit(false);
final Future<ActorSelection> ret = TransactionReadyReplyMapper.transform(messageFuture, actorContext,
if (failure != null) {
LOG.error("Failed to prepare transaction {} on backend", transaction.getIdentifier(), failure);
transactionAborted(transaction);
- } else if (CommitTransactionReply.SERIALIZABLE_CLASS.isInstance(message)) {
+ } else if (CommitTransactionReply.isSerializedType(message)) {
LOG.debug("Transaction {} committed successfully", transaction.getIdentifier());
transactionCommitted(transaction);
} else {
throw new UnsupportedOperationException();
}
- protected void transactionAborted(SnapshotBackedWriteTransaction<TransactionIdentifier> transaction) {
+ protected void transactionAborted(SnapshotBackedWriteTransaction<TransactionIdentifier> aborted) {
}
- protected void transactionCommitted(SnapshotBackedWriteTransaction<TransactionIdentifier> transaction) {
+ protected void transactionCommitted(SnapshotBackedWriteTransaction<TransactionIdentifier> comitted) {
}
}