*/
package org.opendaylight.controller.cluster.datastore;
+import static java.util.Objects.requireNonNull;
+
import akka.actor.ActorSelection;
import akka.dispatch.Futures;
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 java.util.Optional;
+import java.util.SortedSet;
+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;
-import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
-import org.opendaylight.controller.sal.core.spi.data.SnapshotBackedWriteTransaction;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
+import org.opendaylight.controller.cluster.datastore.utils.ActorUtils;
+import org.opendaylight.mdsal.common.api.CommitInfo;
+import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
+import org.opendaylight.mdsal.dom.spi.store.SnapshotBackedWriteTransaction;
+import org.opendaylight.yangtools.yang.common.Empty;
+import org.opendaylight.yangtools.yang.data.tree.api.DataTreeModification;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.concurrent.Future;
private final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction;
private final DataTreeModification modification;
- private final ActorContext actorContext;
+ private final ActorUtils actorUtils;
private final ActorSelection leader;
- private Exception operationError;
-
- protected LocalThreePhaseCommitCohort(final ActorContext actorContext, final ActorSelection leader,
- 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);
+ private final Exception operationError;
+
+ protected LocalThreePhaseCommitCohort(final ActorUtils actorUtils, final ActorSelection leader,
+ final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction,
+ final DataTreeModification modification,
+ final Exception operationError) {
+ this.actorUtils = requireNonNull(actorUtils);
+ this.leader = requireNonNull(leader);
+ this.transaction = requireNonNull(transaction);
+ this.modification = requireNonNull(modification);
+ this.operationError = operationError;
}
- protected LocalThreePhaseCommitCohort(final ActorContext actorContext, final ActorSelection leader,
+ protected LocalThreePhaseCommitCohort(final ActorUtils actorUtils, final ActorSelection leader,
final SnapshotBackedWriteTransaction<TransactionIdentifier> transaction, final Exception operationError) {
- this.actorContext = Preconditions.checkNotNull(actorContext);
- this.leader = Preconditions.checkNotNull(leader);
- this.transaction = Preconditions.checkNotNull(transaction);
- this.operationError = Preconditions.checkNotNull(operationError);
- this.modification = null;
+ this.actorUtils = requireNonNull(actorUtils);
+ this.leader = requireNonNull(leader);
+ this.transaction = requireNonNull(transaction);
+ this.operationError = requireNonNull(operationError);
+ modification = null;
}
- private Future<Object> initiateCommit(final boolean immediate) {
- if(operationError != null) {
+ private Future<Object> initiateCommit(final boolean immediate,
+ final Optional<SortedSet<String>> participatingShardNames) {
+ if (operationError != null) {
return Futures.failed(operationError);
}
- final ReadyLocalTransaction message = new ReadyLocalTransaction(transaction.getIdentifier().toString(),
- modification, immediate);
- return actorContext.executeOperationAsync(leader, message, actorContext.getTransactionCommitOperationTimeout());
- }
-
- void setOperationError(Exception operationError) {
- this.operationError = operationError;
+ final ReadyLocalTransaction message = new ReadyLocalTransaction(transaction.getIdentifier(),
+ modification, immediate, participatingShardNames);
+ return actorUtils.executeOperationAsync(leader, message, actorUtils.getTransactionCommitOperationTimeout());
}
- Future<ActorSelection> initiateCoordinatedCommit() {
- final Future<Object> messageFuture = initiateCommit(false);
- final Future<ActorSelection> ret = TransactionReadyReplyMapper.transform(messageFuture, actorContext,
+ Future<ActorSelection> initiateCoordinatedCommit(final Optional<SortedSet<String>> participatingShardNames) {
+ final Future<Object> messageFuture = initiateCommit(false, participatingShardNames);
+ final Future<ActorSelection> ret = TransactionReadyReplyMapper.transform(messageFuture, actorUtils,
transaction.getIdentifier());
ret.onComplete(new OnComplete<ActorSelection>() {
@Override
- public void onComplete(final Throwable failure, final ActorSelection success) throws Throwable {
+ public void onComplete(final Throwable failure, final ActorSelection success) {
if (failure != null) {
- LOG.info("Failed to prepare transaction {} on backend", transaction.getIdentifier(), failure);
+ LOG.warn("Failed to prepare transaction {} on backend", transaction.getIdentifier(), failure);
transactionAborted(transaction);
return;
}
LOG.debug("Transaction {} resolved to actor {}", transaction.getIdentifier(), success);
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
return ret;
}
Future<Object> initiateDirectCommit() {
- final Future<Object> messageFuture = initiateCommit(true);
- messageFuture.onComplete(new OnComplete<Object>() {
+ final Future<Object> messageFuture = initiateCommit(true, Optional.empty());
+ messageFuture.onComplete(new OnComplete<>() {
@Override
- public void onComplete(final Throwable failure, final Object message) throws Throwable {
+ public void onComplete(final Throwable failure, final Object message) {
if (failure != null) {
- LOG.error("Failed to prepare transaction {} on backend", transaction.getIdentifier(), failure);
+ LOG.warn("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 {
- LOG.error("Transaction {} resulted in unhandled message type {}, aborting", message.getClass());
+ LOG.error("Transaction {} resulted in unhandled message type {}, aborting",
+ transaction.getIdentifier(), message.getClass());
transactionAborted(transaction);
}
}
- }, actorContext.getClientDispatcher());
+ }, actorUtils.getClientDispatcher());
return messageFuture;
}
}
@Override
- public final ListenableFuture<Void> preCommit() {
+ public final ListenableFuture<Empty> preCommit() {
// Intended no-op
throw new UnsupportedOperationException();
}
@Override
- public final ListenableFuture<Void> abort() {
+ public final ListenableFuture<Empty> abort() {
// Intended no-op
throw new UnsupportedOperationException();
}
@Override
- public final ListenableFuture<Void> commit() {
+ public final ListenableFuture<CommitInfo> commit() {
// Intended no-op
throw new UnsupportedOperationException();
}
- protected void transactionAborted(SnapshotBackedWriteTransaction<TransactionIdentifier> transaction) {
+ protected void transactionAborted(final SnapshotBackedWriteTransaction<TransactionIdentifier> aborted) {
}
- protected void transactionCommitted(SnapshotBackedWriteTransaction<TransactionIdentifier> transaction) {
+ protected void transactionCommitted(final SnapshotBackedWriteTransaction<TransactionIdentifier> comitted) {
}
}
Code Review / controller.git / blobdiff