Implement DataChangeListener
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / ThreePhaseCommitCohort.java
1 /*
2  * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
3  *
4  * This program and the accompanying materials are made available under the
5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6  * and is available at http://www.eclipse.org/legal/epl-v10.html
7  */
8
9 package org.opendaylight.controller.cluster.datastore;
10
11 import akka.actor.ActorRef;
12 import akka.actor.PoisonPill;
13 import akka.actor.Props;
14 import akka.event.Logging;
15 import akka.event.LoggingAdapter;
16 import akka.japi.Creator;
17 import com.google.common.util.concurrent.ListenableFuture;
18 import org.opendaylight.controller.cluster.datastore.messages.AbortTransaction;
19 import org.opendaylight.controller.cluster.datastore.messages.AbortTransactionReply;
20 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransaction;
21 import org.opendaylight.controller.cluster.datastore.messages.CanCommitTransactionReply;
22 import org.opendaylight.controller.cluster.datastore.messages.CommitTransaction;
23 import org.opendaylight.controller.cluster.datastore.messages.ForwardedCommitTransaction;
24 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransaction;
25 import org.opendaylight.controller.cluster.datastore.messages.PreCommitTransactionReply;
26 import org.opendaylight.controller.cluster.datastore.modification.CompositeModification;
27 import org.opendaylight.controller.sal.core.spi.data.DOMStoreThreePhaseCommitCohort;
28
29 import java.util.concurrent.ExecutionException;
30
31 public class ThreePhaseCommitCohort extends AbstractUntypedActor {
32     private final DOMStoreThreePhaseCommitCohort cohort;
33     private final ActorRef shardActor;
34     private final CompositeModification modification;
35
36     public ThreePhaseCommitCohort(DOMStoreThreePhaseCommitCohort cohort,
37         ActorRef shardActor, CompositeModification modification) {
38
39         this.cohort = cohort;
40         this.shardActor = shardActor;
41         this.modification = modification;
42     }
43
44     private final LoggingAdapter log =
45         Logging.getLogger(getContext().system(), this);
46
47     public static Props props(final DOMStoreThreePhaseCommitCohort cohort,
48         final ActorRef shardActor, final CompositeModification modification) {
49         return Props.create(new Creator<ThreePhaseCommitCohort>() {
50             @Override
51             public ThreePhaseCommitCohort create() throws Exception {
52                 return new ThreePhaseCommitCohort(cohort, shardActor,
53                     modification);
54             }
55         });
56     }
57
58
59     @Override
60     public void handleReceive(Object message) throws Exception {
61         if (message instanceof CanCommitTransaction) {
62             canCommit((CanCommitTransaction) message);
63         } else if (message instanceof PreCommitTransaction) {
64             preCommit((PreCommitTransaction) message);
65         } else if (message instanceof CommitTransaction) {
66             commit((CommitTransaction) message);
67         } else if (message instanceof AbortTransaction) {
68             abort((AbortTransaction) message);
69         }
70     }
71
72     private void abort(AbortTransaction message) {
73         final ListenableFuture<Void> future = cohort.abort();
74         final ActorRef sender = getSender();
75         final ActorRef self = getSelf();
76
77         future.addListener(new Runnable() {
78             @Override
79             public void run() {
80                 try {
81                     future.get();
82                     sender.tell(new AbortTransactionReply(), self);
83                 } catch (InterruptedException | ExecutionException e) {
84                     log.error(e, "An exception happened when aborting");
85                 }
86             }
87         }, getContext().dispatcher());
88     }
89
90     private void commit(CommitTransaction message) {
91         // Forward the commit to the shard
92         log.debug("Forward commit transaction to Shard {} ", shardActor);
93         shardActor.forward(new ForwardedCommitTransaction(cohort, modification),
94             getContext());
95
96         getContext().parent().tell(PoisonPill.getInstance(), getSelf());
97
98     }
99
100     private void preCommit(PreCommitTransaction message) {
101         final ListenableFuture<Void> future = cohort.preCommit();
102         final ActorRef sender = getSender();
103         final ActorRef self = getSelf();
104
105         future.addListener(new Runnable() {
106             @Override
107             public void run() {
108                 try {
109                     future.get();
110                     sender.tell(new PreCommitTransactionReply(), self);
111                 } catch (InterruptedException | ExecutionException e) {
112                     log.error(e, "An exception happened when preCommitting");
113                 }
114             }
115         }, getContext().dispatcher());
116
117     }
118
119     private void canCommit(CanCommitTransaction message) {
120         final ListenableFuture<Boolean> future = cohort.canCommit();
121         final ActorRef sender = getSender();
122         final ActorRef self = getSelf();
123
124         future.addListener(new Runnable() {
125             @Override
126             public void run() {
127                 try {
128                     Boolean canCommit = future.get();
129                     sender.tell(new CanCommitTransactionReply(canCommit), self);
130                 } catch (InterruptedException | ExecutionException e) {
131                     log.error(e, "An exception happened when aborting");
132                 }
133             }
134         }, getContext().dispatcher());
135
136     }
137 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.