Use odlparent for archetype to inherit the version
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / databroker / actors / dds / AbstractProxyTransaction.java
1 /*
2  * Copyright (c) 2016 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 package org.opendaylight.controller.cluster.databroker.actors.dds;
9
10 import akka.actor.ActorRef;
11 import com.google.common.base.Optional;
12 import com.google.common.base.Preconditions;
13 import com.google.common.base.Verify;
14 import com.google.common.util.concurrent.CheckedFuture;
15 import com.google.common.util.concurrent.ListenableFuture;
16 import com.google.common.util.concurrent.SettableFuture;
17 import java.util.function.Consumer;
18 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
19 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
20 import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
21 import org.opendaylight.controller.cluster.access.commands.TransactionCommitSuccess;
22 import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
23 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
24 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess;
25 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
26 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
27 import org.opendaylight.controller.cluster.access.concepts.Response;
28 import org.opendaylight.controller.cluster.access.concepts.TransactionIdentifier;
29 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
30 import org.opendaylight.yangtools.concepts.Identifiable;
31 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
32 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
33
34 /**
35  * Class translating transaction operations towards a particular backend shard.
36  *
37  * <p>
38  * This class is not safe to access from multiple application threads, as is usual for transactions. Internal state
39  * transitions coming from interactions with backend are expected to be thread-safe.
40  *
41  * <p>
42  * This class interacts with the queueing mechanism in ClientActorBehavior, hence once we arrive at a decision
43  * to use either a local or remote implementation, we are stuck with it. We can re-evaluate on the next transaction.
44  *
45  * @author Robert Varga
46  */
47 abstract class AbstractProxyTransaction implements Identifiable<TransactionIdentifier> {
48     private final DistributedDataStoreClientBehavior client;
49
50     private long sequence;
51     private boolean sealed;
52
53     AbstractProxyTransaction(final DistributedDataStoreClientBehavior client) {
54         this.client = Preconditions.checkNotNull(client);
55     }
56
57     final ActorRef localActor() {
58         return client.self();
59     }
60
61     final long nextSequence() {
62         return sequence++;
63     }
64
65     final void delete(final YangInstanceIdentifier path) {
66         checkSealed();
67         doDelete(path);
68     }
69
70     final void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
71         checkSealed();
72         doMerge(path, data);
73     }
74
75     final void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
76         checkSealed();
77         doWrite(path, data);
78     }
79
80     final CheckedFuture<Boolean, ReadFailedException> exists(final YangInstanceIdentifier path) {
81         checkSealed();
82         return doExists(path);
83     }
84
85     final CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> read(final YangInstanceIdentifier path) {
86         checkSealed();
87         return doRead(path);
88     }
89
90     final void sendRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> completer) {
91         client.sendRequest(request, completer);
92     }
93
94     /**
95      * Seals this transaction when ready.
96      */
97     final void seal() {
98         checkSealed();
99         doSeal();
100         sealed = true;
101     }
102
103     private void checkSealed() {
104         Preconditions.checkState(sealed, "Transaction %s has not been sealed yet", getIdentifier());
105     }
106
107     /**
108      * Abort this transaction. This is invoked only for read-only transactions and will result in an explicit message
109      * being sent to the backend.
110      */
111     final void abort() {
112         checkSealed();
113         doAbort();
114     }
115
116     void abort(final VotingFuture<Void> ret) {
117         checkSealed();
118
119         sendRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), t -> {
120             if (t instanceof TransactionAbortSuccess) {
121                 ret.voteYes();
122             } else if (t instanceof RequestFailure) {
123                 ret.voteNo(((RequestFailure<?, ?>) t).getCause());
124             } else {
125                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
126             }
127         });
128     }
129
130     /**
131      * Commit this transaction, possibly in a coordinated fashion.
132      *
133      * @param coordinated True if this transaction should be coordinated across multiple participants.
134      * @return Future completion
135      */
136     final ListenableFuture<Boolean> directCommit() {
137         checkSealed();
138
139         final SettableFuture<Boolean> ret = SettableFuture.create();
140         sendRequest(Verify.verifyNotNull(doCommit(false)), t -> {
141             if (t instanceof TransactionCommitSuccess) {
142                 ret.set(Boolean.TRUE);
143             } else if (t instanceof RequestFailure) {
144                 ret.setException(((RequestFailure<?, ?>) t).getCause());
145             } else {
146                 ret.setException(new IllegalStateException("Unhandled response " + t.getClass()));
147             }
148         });
149         return ret;
150     }
151
152     void canCommit(final VotingFuture<?> ret) {
153         checkSealed();
154
155         sendRequest(Verify.verifyNotNull(doCommit(true)), t -> {
156             if (t instanceof TransactionCanCommitSuccess) {
157                 ret.voteYes();
158             } else if (t instanceof RequestFailure) {
159                 ret.voteNo(((RequestFailure<?, ?>) t).getCause());
160             } else {
161                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
162             }
163         });
164     }
165
166     void preCommit(final VotingFuture<?> ret) {
167         checkSealed();
168
169         sendRequest(new TransactionPreCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> {
170             if (t instanceof TransactionPreCommitSuccess) {
171                 ret.voteYes();
172             } else if (t instanceof RequestFailure) {
173                 ret.voteNo(((RequestFailure<?, ?>) t).getCause());
174             } else {
175                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
176             }
177         });
178     }
179
180     void doCommit(final VotingFuture<?> ret) {
181         checkSealed();
182
183         sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> {
184             if (t instanceof TransactionCommitSuccess) {
185                 ret.voteYes();
186             } else if (t instanceof RequestFailure) {
187                 ret.voteNo(((RequestFailure<?, ?>) t).getCause());
188             } else {
189                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
190             }
191         });
192     }
193
194     abstract TransactionRequest<?> doCommit(boolean coordinated);
195
196     abstract void doDelete(final YangInstanceIdentifier path);
197
198     abstract void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data);
199
200     abstract void doWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data);
201
202     abstract CheckedFuture<Boolean, ReadFailedException> doExists(final YangInstanceIdentifier path);
203
204     abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(
205             final YangInstanceIdentifier path);
206
207     abstract void doSeal();
208
209     abstract void doAbort();
210 }