+
+ @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);
+ }
+ }
+
+ 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");
+ }
+ }
+ }, getContext().dispatcher());
+
+ }