X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FThreePhaseCommitCohort.java;h=25705bff418740c873af8334091b3961612c3f1e;hp=61baf1ab64421e04f76d52ec684709ca33f38d25;hb=d206d27042eef2185c875f85cf6eac61a1bd77c4;hpb=e55942fd1fcc4748197146b81f7497256de22a94

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 61baf1ab64..25705bff41 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
@@ -9,11 +9,13 @@
 package org.opendaylight.controller.cluster.datastore;
 
 import akka.actor.ActorRef;
+import akka.actor.PoisonPill;
 import akka.actor.Props;
-import akka.actor.UntypedActor;
 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;
@@ -26,103 +28,120 @@ import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransacti
 import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
 
-import java.util.concurrent.ExecutionException;
-
-public class ThreePhaseCommitCohort extends UntypedActor{
-  private final DOMStoreThreePhaseCommitCohort cohort;
-  private final ActorRef shardActor;
-  private final CompositeModification modification;
-
-  public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort, ActorRef shardActor, CompositeModification modification) {
-    this.cohort = cohort;
-    this.shardActor = shardActor;
-    this.modification = modification;
-  }
-
-  private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);
-
-  public static Props props(final DOMStoreThreePhaseCommitCohort cohort, final ActorRef shardActor, final CompositeModification modification) {
-    return Props.create(new Creator<ThreePhaseCommitCohort>(){
-      @Override
-      public ThreePhaseCommitCohort create() throws Exception {
-        return new ThreePhaseCommitCohort(cohort, shardActor, modification);
-      }
-    });
-  }
-
-  @Override
-  public void onReceive(Object message) throws Exception {
-    if(message instanceof CanCommitTransaction){
-      canCommit((CanCommitTransaction) message);
-    } else if(message instanceof PreCommitTransaction) {
-      preCommit((PreCommitTransaction) message);
-    } else if(message instanceof CommitTransaction){
-      commit((CommitTransaction) message);
-    } else if (message instanceof AbortTransaction){
-      abort((AbortTransaction) message);
+public class ThreePhaseCommitCohort extends AbstractUntypedActor {
+    private final DOMStoreThreePhaseCommitCohort cohort;
+    private final ActorRef shardActor;
+    private final CompositeModification modification;
+
+    public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort,
+        ActorRef shardActor, CompositeModification modification) {
+
+        this.cohort = cohort;
+        this.shardActor = shardActor;
+        this.modification = modification;
     }
-  }
-
-  private void abort(AbortTransaction message) {
-    final ListenableFuture<Void> future = cohort.abort();
-    final ActorRef sender = getSender();
-    final ActorRef self = getSelf();
-
-    future.addListener(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          future.get();
-          sender.tell(new AbortTransactionReply(), self);
-        } catch (InterruptedException | ExecutionException e) {
-          log.error(e, "An exception happened when aborting");
-        }
-      }
-    }, getContext().dispatcher());
-  }
-
-  private void commit(CommitTransaction message) {
-    // Forward the commit to the shard
-    log.info("Commit transaction now + " + shardActor);
-    shardActor.forward(new ForwardedCommitTransaction(cohort, modification), getContext());
-
-  }
-
-  private void preCommit(PreCommitTransaction message) {
-    final ListenableFuture<Void> future = cohort.preCommit();
-    final ActorRef sender = getSender();
-    final ActorRef self = getSelf();
-
-    future.addListener(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          future.get();
-          sender.tell(new PreCommitTransactionReply(), self);
-        } catch (InterruptedException | ExecutionException e) {
-          log.error(e, "An exception happened when preCommitting");
-        }
-      }
-    }, getContext().dispatcher());
-
-  }
-
-  private void canCommit(CanCommitTransaction message) {
-    final ListenableFuture<Boolean> future = cohort.canCommit();
-    final ActorRef sender = getSender();
-    final ActorRef self = getSelf();
-
-    future.addListener(new Runnable() {
-      @Override
-      public void run() {
-        try {
-          Boolean canCommit = future.get();
-          sender.tell(new CanCommitTransactionReply(canCommit), self);
-        } catch (InterruptedException | ExecutionException e) {
-          log.error(e, "An exception happened when aborting");
+
+    private final LoggingAdapter log =
+        Logging.getLogger(getContext().system(), this);
+
+    public static Props props(final DOMStoreThreePhaseCommitCohort cohort,
+        final ActorRef shardActor, final CompositeModification modification) {
+        return Props.create(new Creator<ThreePhaseCommitCohort>() {
+            @Override
+            public ThreePhaseCommitCohort create() throws Exception {
+                return new ThreePhaseCommitCohort(cohort, shardActor,
+                    modification);
+            }
+        });
+    }
+
+
+    @Override
+    public void handleReceive(Object message) throws Exception {
+        if (message.getClass()
+            .equals(CanCommitTransaction.SERIALIZABLE_CLASS)) {
+            canCommit(new CanCommitTransaction());
+        } else if (message.getClass()
+            .equals(PreCommitTransaction.SERIALIZABLE_CLASS)) {
+            preCommit(new PreCommitTransaction());
+        } else if (message.getClass()
+            .equals(CommitTransaction.SERIALIZABLE_CLASS)) {
+            commit(new CommitTransaction());
+        } else if (message.getClass()
+            .equals(AbortTransaction.SERIALIZABLE_CLASS)) {
+            abort(new AbortTransaction());
+        } else {
+            unknownMessage(message);
         }
-      }
-    }, getContext().dispatcher());
+    }
+
+    private void abort(AbortTransaction message) {
+        final ListenableFuture<Void> future = cohort.abort();
+        final ActorRef sender = getSender();
+        final ActorRef self = getSelf();
+
+        Futures.addCallback(future, new FutureCallback<Void>() {
+            public void onSuccess(Void v) {
+                sender
+                    .tell(new AbortTransactionReply().toSerializable(),
+                        self);
+            }
+
+            public void onFailure(Throwable t) {
+                LOG.error(t, "An exception happened during abort");
+                sender
+                    .tell(new akka.actor.Status.Failure(t), getSelf());
+            }
+        });
+    }
 
-  }
+    private void commit(CommitTransaction message) {
+        // Forward the commit to the shard
+        log.debug("Forward commit transaction to Shard {} ", shardActor);
+        shardActor.forward(new ForwardedCommitTransaction(cohort, modification),
+            getContext());
+
+        getContext().parent().tell(PoisonPill.getInstance(), getSelf());
+
+    }
+
+    private void preCommit(PreCommitTransaction message) {
+        final ListenableFuture<Void> future = cohort.preCommit();
+        final ActorRef sender = getSender();
+        final ActorRef self = getSelf();
+        Futures.addCallback(future, new FutureCallback<Void>() {
+            public void onSuccess(Void v) {
+                sender
+                    .tell(new PreCommitTransactionReply().toSerializable(),
+                        self);
+            }
+
+            public void onFailure(Throwable t) {
+                LOG.error(t, "An exception happened during pre-commit");
+                sender
+                    .tell(new akka.actor.Status.Failure(t), getSelf());
+            }
+        });
+
+    }
+
+    private void canCommit(CanCommitTransaction message) {
+        final ListenableFuture<Boolean> future = cohort.canCommit();
+        final ActorRef sender = getSender();
+        final ActorRef self = getSelf();
+        Futures.addCallback(future, new FutureCallback<Boolean>() {
+            public void onSuccess(Boolean canCommit) {
+                sender.tell(new CanCommitTransactionReply(canCommit)
+                    .toSerializable(), self);
+            }
+
+            public void onFailure(Throwable t) {
+                LOG.error(t, "An exception happened during canCommit");
+                sender
+                    .tell(new akka.actor.Status.Failure(t), getSelf());
+            }
+        });
+
+
+    }
 }