2 * Copyright (c) 2017 Pantheon Technologies s.r.o. and others. All rights reserved.
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
8 package org.opendaylight.controller.cluster.databroker.actors.dds;
10 import static org.junit.Assert.assertEquals;
11 import static org.opendaylight.controller.cluster.databroker.actors.dds.TestUtils.CLIENT_ID;
12 import static org.opendaylight.controller.cluster.databroker.actors.dds.TestUtils.HISTORY_ID;
13 import static org.opendaylight.controller.cluster.databroker.actors.dds.TestUtils.TRANSACTION_ID;
14 import static org.opendaylight.controller.cluster.databroker.actors.dds.TestUtils.assertOperationThrowsException;
15 import static org.opendaylight.controller.cluster.databroker.actors.dds.TestUtils.getWithTimeout;
17 import akka.actor.ActorSystem;
18 import akka.testkit.TestProbe;
19 import akka.testkit.javadsl.TestKit;
20 import com.google.common.primitives.UnsignedLong;
21 import com.google.common.util.concurrent.ListenableFuture;
22 import java.util.ArrayList;
23 import java.util.Collection;
24 import java.util.List;
25 import java.util.Optional;
26 import java.util.concurrent.ExecutionException;
27 import java.util.function.Consumer;
28 import java.util.function.Function;
29 import java.util.stream.Collectors;
30 import org.junit.After;
31 import org.junit.Before;
32 import org.junit.Test;
33 import org.junit.runner.RunWith;
34 import org.mockito.Mock;
35 import org.mockito.junit.MockitoJUnitRunner;
36 import org.opendaylight.controller.cluster.access.ABIVersion;
37 import org.opendaylight.controller.cluster.access.client.AbstractClientConnection;
38 import org.opendaylight.controller.cluster.access.client.AccessClientUtil;
39 import org.opendaylight.controller.cluster.access.client.ClientActorContext;
40 import org.opendaylight.controller.cluster.access.commands.ModifyTransactionRequest;
41 import org.opendaylight.controller.cluster.access.commands.PersistenceProtocol;
42 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
43 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
44 import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
45 import org.opendaylight.controller.cluster.access.commands.TransactionCommitSuccess;
46 import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
47 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
48 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess;
49 import org.opendaylight.controller.cluster.access.concepts.RequestSuccess;
50 import org.opendaylight.controller.cluster.access.concepts.RuntimeRequestException;
52 @RunWith(MockitoJUnitRunner.StrictStubs.class)
53 public class ClientTransactionCommitCohortTest {
55 private static final String PERSISTENCE_ID = "per-1";
56 private static final int TRANSACTIONS = 3;
59 private AbstractClientHistory history;
60 private ActorSystem system;
61 private List<TransactionTester<RemoteProxyTransaction>> transactions;
62 private ClientTransactionCommitCohort cohort;
66 system = ActorSystem.apply();
67 final TestProbe clientContextProbe = new TestProbe(system, "clientContext");
68 final ClientActorContext context =
69 AccessClientUtil.createClientActorContext(system, clientContextProbe.ref(), CLIENT_ID, PERSISTENCE_ID);
70 transactions = new ArrayList<>();
71 for (int i = 0; i < TRANSACTIONS; i++) {
72 transactions.add(createTransactionTester(new TestProbe(system, "backend" + i), context, history));
74 final Collection<AbstractProxyTransaction> proxies = transactions.stream()
75 .map(TransactionTester::getTransaction)
76 .collect(Collectors.toList());
77 proxies.forEach(AbstractProxyTransaction::seal);
78 cohort = new ClientTransactionCommitCohort(history, TRANSACTION_ID, proxies);
82 public void tearDown() {
83 TestKit.shutdownActorSystem(system);
87 public void testCanCommit() throws Exception {
88 testOpSuccess(ClientTransactionCommitCohort::canCommit, this::expectCanCommit,
89 this::replyCanCommitSuccess, Boolean.TRUE);
93 public void testCanCommitFail() throws Exception {
94 testOpFail(ClientTransactionCommitCohort::canCommit, this::expectCanCommit, this::replyCanCommitSuccess);
98 public void testPreCommit() throws Exception {
99 testOpSuccess(ClientTransactionCommitCohort::preCommit, this::expectPreCommit, this::replyPreCommitSuccess,
104 public void testPreCommitFail() throws Exception {
105 testOpFail(ClientTransactionCommitCohort::preCommit, this::expectPreCommit, this::replyPreCommitSuccess);
109 public void testCommit() throws Exception {
110 testOpSuccess(ClientTransactionCommitCohort::commit, this::expectCommit, this::replyCommitSuccess, null);
114 public void testCommitFail() throws Exception {
115 testOpFail(ClientTransactionCommitCohort::commit, this::expectCommit, this::replyCommitSuccess);
119 public void testAbort() throws Exception {
120 testOpSuccess(ClientTransactionCommitCohort::abort, this::expectAbort, this::replyAbortSuccess, null);
124 public void testAbortFail() throws Exception {
125 testOpFail(ClientTransactionCommitCohort::abort, this::expectAbort, this::replyAbortSuccess);
128 private void expectCanCommit(final TransactionTester<RemoteProxyTransaction> tester) {
129 final ModifyTransactionRequest request = tester.expectTransactionRequest(ModifyTransactionRequest.class);
130 assertEquals(Optional.of(PersistenceProtocol.THREE_PHASE), request.getPersistenceProtocol());
133 void expectPreCommit(final TransactionTester<?> tester) {
134 tester.expectTransactionRequest(TransactionPreCommitRequest.class);
137 void expectCommit(final TransactionTester<?> tester) {
138 tester.expectTransactionRequest(TransactionDoCommitRequest.class);
141 void expectAbort(final TransactionTester<?> tester) {
142 tester.expectTransactionRequest(TransactionAbortRequest.class);
145 void replyCanCommitSuccess(final TransactionTester<?> tester) {
146 final RequestSuccess<?, ?> success = new TransactionCanCommitSuccess(tester.getTransaction().getIdentifier(),
147 tester.getLastReceivedMessage().getSequence());
148 tester.replySuccess(success);
151 void replyPreCommitSuccess(final TransactionTester<?> tester) {
152 final RequestSuccess<?, ?> success = new TransactionPreCommitSuccess(tester.getTransaction().getIdentifier(),
153 tester.getLastReceivedMessage().getSequence());
154 tester.replySuccess(success);
157 void replyCommitSuccess(final TransactionTester<?> tester) {
158 final RequestSuccess<?, ?> success = new TransactionCommitSuccess(tester.getTransaction().getIdentifier(),
159 tester.getLastReceivedMessage().getSequence());
160 tester.replySuccess(success);
163 void replyAbortSuccess(final TransactionTester<?> tester) {
164 final RequestSuccess<?, ?> success = new TransactionAbortSuccess(tester.getTransaction().getIdentifier(),
165 tester.getLastReceivedMessage().getSequence());
166 tester.replySuccess(success);
169 private static TransactionTester<RemoteProxyTransaction> createTransactionTester(final TestProbe backendProbe,
170 final ClientActorContext context,
171 final AbstractClientHistory history) {
172 final ShardBackendInfo backend = new ShardBackendInfo(backendProbe.ref(), 0L, ABIVersion.BORON,
173 "default", UnsignedLong.ZERO, Optional.empty(), 3);
174 final AbstractClientConnection<ShardBackendInfo> connection =
175 AccessClientUtil.createConnectedConnection(context, 0L, backend);
176 final ProxyHistory proxyHistory = ProxyHistory.createClient(history, connection, HISTORY_ID);
177 final RemoteProxyTransaction transaction =
178 new RemoteProxyTransaction(proxyHistory, TRANSACTION_ID, false, false, false);
179 return new TransactionTester<>(transaction, connection, backendProbe);
182 private static <T extends TransactionTester<?>> void replySuccess(final Collection<T> transactions,
183 final Consumer<T> expect, final Consumer<T> reply) {
184 for (final T transaction : transactions) {
185 expect.accept(transaction);
186 reply.accept(transaction);
191 * Test operation success. Invokes given operation, which initiates message to the backend.
192 * Received message is checked by expectFunction. Then replyFunction is invoked. Expected result is compared
193 * to the operation future result.
195 * @param operation operation
196 * @param expectFunction expected message check
197 * @param replyFunction response function
198 * @param expectedResult expected operation result
200 * @throws Exception unexpected exception
202 private <T> void testOpSuccess(final Function<ClientTransactionCommitCohort, ListenableFuture<T>> operation,
203 final Consumer<TransactionTester<RemoteProxyTransaction>> expectFunction,
204 final Consumer<TransactionTester<RemoteProxyTransaction>> replyFunction,
205 final T expectedResult) throws Exception {
206 final ListenableFuture<T> result = operation.apply(cohort);
207 replySuccess(transactions, expectFunction, replyFunction);
208 assertEquals(expectedResult, getWithTimeout(result));
212 * Test operation failure. Invokes given operation, which initiates message to the backend.
213 * Received message is checked by expectFunction. Then replyFunction is invoked. One of the transactions in
214 * cohort receives failure response.
216 * @param operation operation
217 * @param expectFunction expected message check
218 * @param replyFunction response function
220 * @throws Exception unexpected exception
222 private <T> void testOpFail(final Function<ClientTransactionCommitCohort, ListenableFuture<T>> operation,
223 final Consumer<TransactionTester<RemoteProxyTransaction>> expectFunction,
224 final Consumer<TransactionTester<RemoteProxyTransaction>> replyFunction) throws Exception {
225 final ListenableFuture<T> canCommit = operation.apply(cohort);
226 //reply success to all except last transaction
227 replySuccess(transactions.subList(0, transactions.size() - 1), expectFunction, replyFunction);
228 //reply fail to last transaction
229 final TransactionTester<RemoteProxyTransaction> last = transactions.get(transactions.size() - 1);
230 expectFunction.accept(last);
231 final RuntimeException e = new RuntimeException();
232 final RuntimeRequestException cause = new RuntimeRequestException("fail", e);
233 last.replyFailure(cause);
235 final ExecutionException exception =
236 assertOperationThrowsException(() -> getWithTimeout(canCommit), ExecutionException.class);
237 assertEquals(e, exception.getCause());