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%2FThreePhaseCommitCohort.java;h=d0c29294cbea9b992644817436ad21fb5f81d46b;hb=9cd4e7995210f8381892004373acc71c8b3ae7af;hp=25705bff418740c873af8334091b3961612c3f1e;hpb=dcd30725cd5b176da433d6217bb6e0ea00edd96c;p=controller.git diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java index 25705bff41..d0c29294cb 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/ThreePhaseCommitCohort.java @@ -14,9 +14,11 @@ import akka.actor.Props; import akka.event.Logging; import akka.event.LoggingAdapter; import akka.japi.Creator; + import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; + import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction; import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply; import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction; @@ -46,16 +48,9 @@ public class ThreePhaseCommitCohort extends AbstractUntypedActor { public static Props props(final DOMStoreThreePhaseCommitCohort cohort, final ActorRef shardActor, final CompositeModification modification) { - return Props.create(new Creator() { - @Override - public ThreePhaseCommitCohort create() throws Exception { - return new ThreePhaseCommitCohort(cohort, shardActor, - modification); - } - }); + return Props.create(new ThreePhaseCommitCohortCreator(cohort, shardActor, modification)); } - @Override public void handleReceive(Object message) throws Exception { if (message.getClass() @@ -81,16 +76,18 @@ public class ThreePhaseCommitCohort extends AbstractUntypedActor { final ActorRef self = getSelf(); Futures.addCallback(future, new FutureCallback() { + @Override public void onSuccess(Void v) { sender .tell(new AbortTransactionReply().toSerializable(), self); } + @Override public void onFailure(Throwable t) { LOG.error(t, "An exception happened during abort"); sender - .tell(new akka.actor.Status.Failure(t), getSelf()); + .tell(new akka.actor.Status.Failure(t), self); } }); } @@ -110,16 +107,18 @@ public class ThreePhaseCommitCohort extends AbstractUntypedActor { final ActorRef sender = getSender(); final ActorRef self = getSelf(); Futures.addCallback(future, new FutureCallback() { + @Override public void onSuccess(Void v) { sender .tell(new PreCommitTransactionReply().toSerializable(), self); } + @Override public void onFailure(Throwable t) { LOG.error(t, "An exception happened during pre-commit"); sender - .tell(new akka.actor.Status.Failure(t), getSelf()); + .tell(new akka.actor.Status.Failure(t), self); } }); @@ -130,18 +129,36 @@ public class ThreePhaseCommitCohort extends AbstractUntypedActor { final ActorRef sender = getSender(); final ActorRef self = getSelf(); Futures.addCallback(future, new FutureCallback() { + @Override public void onSuccess(Boolean canCommit) { sender.tell(new CanCommitTransactionReply(canCommit) .toSerializable(), self); } + @Override public void onFailure(Throwable t) { LOG.error(t, "An exception happened during canCommit"); sender - .tell(new akka.actor.Status.Failure(t), getSelf()); + .tell(new akka.actor.Status.Failure(t), self); } }); + } + + private static class ThreePhaseCommitCohortCreator implements Creator { + final DOMStoreThreePhaseCommitCohort cohort; + final ActorRef shardActor; + final CompositeModification modification; + ThreePhaseCommitCohortCreator(DOMStoreThreePhaseCommitCohort cohort, + ActorRef shardActor, CompositeModification modification) { + this.cohort = cohort; + this.shardActor = shardActor; + this.modification = modification; + } + @Override + public ThreePhaseCommitCohort create() throws Exception { + return new ThreePhaseCommitCohort(cohort, shardActor, modification); + } } }