acf08eb1d9e50fffea8dcf7470464cf6fe05a301
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / databroker / ConcurrentDOMDataBrokerTest.java
1 /*
2  * Copyright (c) 2014 Brocade Communications 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;
9
10 import static org.junit.Assert.assertEquals;
11 import static org.junit.Assert.assertNotNull;
12 import static org.junit.Assert.assertSame;
13 import static org.junit.Assert.assertTrue;
14 import static org.junit.Assert.fail;
15 import static org.mockito.Mockito.doAnswer;
16 import static org.mockito.Mockito.doReturn;
17 import static org.mockito.Mockito.inOrder;
18 import static org.mockito.Mockito.mock;
19 import static org.mockito.Mockito.never;
20 import static org.mockito.Mockito.times;
21 import static org.mockito.Mockito.verify;
22
23 import com.google.common.base.Throwables;
24 import com.google.common.collect.ImmutableMap;
25 import com.google.common.util.concurrent.CheckedFuture;
26 import com.google.common.util.concurrent.FluentFuture;
27 import com.google.common.util.concurrent.FutureCallback;
28 import com.google.common.util.concurrent.Futures;
29 import com.google.common.util.concurrent.ListenableFuture;
30 import com.google.common.util.concurrent.MoreExecutors;
31 import com.google.common.util.concurrent.SettableFuture;
32 import com.google.common.util.concurrent.Uninterruptibles;
33 import java.util.ArrayList;
34 import java.util.Arrays;
35 import java.util.Collection;
36 import java.util.List;
37 import java.util.Map;
38 import java.util.concurrent.CountDownLatch;
39 import java.util.concurrent.ExecutionException;
40 import java.util.concurrent.SynchronousQueue;
41 import java.util.concurrent.ThreadPoolExecutor;
42 import java.util.concurrent.TimeUnit;
43 import java.util.concurrent.TimeoutException;
44 import java.util.concurrent.atomic.AtomicReference;
45 import org.junit.After;
46 import org.junit.Before;
47 import org.junit.Test;
48 import org.mockito.InOrder;
49 import org.mockito.stubbing.Answer;
50 import org.opendaylight.controller.cluster.datastore.DistributedDataStore;
51 import org.opendaylight.controller.cluster.datastore.exceptions.NoShardLeaderException;
52 import org.opendaylight.mdsal.common.api.CommitInfo;
53 import org.opendaylight.mdsal.common.api.DataStoreUnavailableException;
54 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
55 import org.opendaylight.mdsal.common.api.TransactionChainListener;
56 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
57 import org.opendaylight.mdsal.dom.api.DOMDataBrokerExtension;
58 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeService;
59 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohort;
60 import org.opendaylight.mdsal.dom.api.DOMDataTreeCommitCohortRegistry;
61 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
62 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadTransaction;
63 import org.opendaylight.mdsal.dom.api.DOMDataTreeReadWriteTransaction;
64 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
65 import org.opendaylight.mdsal.dom.api.DOMTransactionChain;
66 import org.opendaylight.mdsal.dom.broker.TransactionCommitFailedExceptionMapper;
67 import org.opendaylight.mdsal.dom.spi.store.DOMStore;
68 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadTransaction;
69 import org.opendaylight.mdsal.dom.spi.store.DOMStoreReadWriteTransaction;
70 import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
71 import org.opendaylight.mdsal.dom.spi.store.DOMStoreTransactionChain;
72 import org.opendaylight.mdsal.dom.spi.store.DOMStoreWriteTransaction;
73 import org.opendaylight.mdsal.dom.store.inmemory.InMemoryDOMDataStore;
74 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
75 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
76
77 /**
78  * Unit tests for DOMConcurrentDataCommitCoordinator.
79  *
80  * @author Thomas Pantelis
81  */
82 public class ConcurrentDOMDataBrokerTest {
83
84     private final DOMDataTreeWriteTransaction transaction = mock(DOMDataTreeWriteTransaction.class);
85     private final DOMStoreThreePhaseCommitCohort mockCohort1 = mock(DOMStoreThreePhaseCommitCohort.class);
86     private final DOMStoreThreePhaseCommitCohort mockCohort2 = mock(DOMStoreThreePhaseCommitCohort.class);
87     private final ThreadPoolExecutor futureExecutor =
88             new ThreadPoolExecutor(0, 1, 5, TimeUnit.SECONDS, new SynchronousQueue<>());
89     private ConcurrentDOMDataBroker coordinator;
90
91     @Before
92     public void setup() {
93         doReturn("tx").when(transaction).getIdentifier();
94
95         DOMStore store = new InMemoryDOMDataStore("OPER", MoreExecutors.newDirectExecutorService());
96
97         coordinator = new ConcurrentDOMDataBroker(ImmutableMap.of(LogicalDatastoreType.OPERATIONAL, store),
98                 futureExecutor);
99     }
100
101     @After
102     public void tearDown() {
103         futureExecutor.shutdownNow();
104     }
105
106     @Test
107     public void testSuccessfulSubmitAsync() throws Exception {
108         testSuccessfulSubmit(true);
109     }
110
111     @Test
112     public void testSuccessfulSubmitSync() throws Exception {
113         testSuccessfulSubmit(false);
114     }
115
116     private void testSuccessfulSubmit(final boolean doAsync) throws InterruptedException {
117         final CountDownLatch asyncCanCommitContinue = new CountDownLatch(1);
118         Answer<ListenableFuture<Boolean>> asyncCanCommit = invocation -> {
119             final SettableFuture<Boolean> future = SettableFuture.create();
120             if (doAsync) {
121                 new Thread(() -> {
122                     Uninterruptibles.awaitUninterruptibly(asyncCanCommitContinue,
123                             10, TimeUnit.SECONDS);
124                     future.set(true);
125                 }).start();
126             } else {
127                 future.set(true);
128             }
129
130             return future;
131         };
132
133         doAnswer(asyncCanCommit).when(mockCohort1).canCommit();
134         doReturn(Futures.immediateFuture(null)).when(mockCohort1).preCommit();
135         doReturn(Futures.immediateFuture(null)).when(mockCohort1).commit();
136
137         doReturn(Futures.immediateFuture(true)).when(mockCohort2).canCommit();
138         doReturn(Futures.immediateFuture(null)).when(mockCohort2).preCommit();
139         doReturn(Futures.immediateFuture(null)).when(mockCohort2).commit();
140
141         ListenableFuture<? extends CommitInfo> future =
142                 coordinator.commit(transaction, Arrays.asList(mockCohort1, mockCohort2));
143
144         final CountDownLatch doneLatch = new CountDownLatch(1);
145         final AtomicReference<Throwable> caughtEx = new AtomicReference<>();
146         Futures.addCallback(future, new FutureCallback<CommitInfo>() {
147             @Override
148             public void onSuccess(final CommitInfo result) {
149                 doneLatch.countDown();
150             }
151
152             @Override
153             public void onFailure(final Throwable failure) {
154                 caughtEx.set(failure);
155                 doneLatch.countDown();
156             }
157         }, MoreExecutors.directExecutor());
158
159         asyncCanCommitContinue.countDown();
160
161         assertEquals("Submit complete", true, doneLatch.await(5, TimeUnit.SECONDS));
162
163         if (caughtEx.get() != null) {
164             Throwables.throwIfUnchecked(caughtEx.get());
165             throw new RuntimeException(caughtEx.get());
166         }
167
168         assertEquals("Task count", doAsync ? 1 : 0, futureExecutor.getTaskCount());
169
170         InOrder inOrder = inOrder(mockCohort1, mockCohort2);
171         inOrder.verify(mockCohort1).canCommit();
172         inOrder.verify(mockCohort2).canCommit();
173         inOrder.verify(mockCohort1).preCommit();
174         inOrder.verify(mockCohort2).preCommit();
175         inOrder.verify(mockCohort1).commit();
176         inOrder.verify(mockCohort2).commit();
177     }
178
179     @Test
180     public void testSubmitWithNegativeCanCommitResponse() throws Exception {
181         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
182         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
183
184         doReturn(Futures.immediateFuture(false)).when(mockCohort2).canCommit();
185         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
186
187         DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class);
188         doReturn(Futures.immediateFuture(false)).when(mockCohort3).canCommit();
189         doReturn(Futures.immediateFuture(null)).when(mockCohort3).abort();
190
191         ListenableFuture<? extends CommitInfo> future = coordinator.commit(
192                 transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3));
193
194         assertFailure(future, null, mockCohort1, mockCohort2, mockCohort3);
195     }
196
197     private static void assertFailure(final ListenableFuture<?> future,
198             final Exception expCause, final DOMStoreThreePhaseCommitCohort... mockCohorts)
199                     throws Exception {
200         try {
201             future.get(5, TimeUnit.SECONDS);
202             fail("Expected TransactionCommitFailedException");
203         } catch (ExecutionException e) {
204             TransactionCommitFailedException tcf = TransactionCommitFailedExceptionMapper.COMMIT_ERROR_MAPPER.apply(e);
205             if (expCause != null) {
206                 assertSame("Expected cause", expCause.getClass(), tcf.getCause().getClass());
207             }
208
209             InOrder inOrder = inOrder((Object[])mockCohorts);
210             for (DOMStoreThreePhaseCommitCohort c: mockCohorts) {
211                 inOrder.verify(c).abort();
212             }
213         } catch (TimeoutException e) {
214             throw e;
215         }
216     }
217
218     @Test
219     public void testSubmitWithCanCommitException() throws Exception {
220         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
221         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
222
223         IllegalStateException cause = new IllegalStateException("mock");
224         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).canCommit();
225         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
226
227         FluentFuture<? extends CommitInfo> future = coordinator.commit(
228                 transaction, Arrays.asList(mockCohort1, mockCohort2));
229
230         assertFailure(future, cause, mockCohort1, mockCohort2);
231     }
232
233     @Test
234     public void testSubmitWithCanCommitDataStoreUnavailableException() throws Exception {
235         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
236         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
237         NoShardLeaderException rootCause = new NoShardLeaderException("mock");
238         DataStoreUnavailableException cause = new DataStoreUnavailableException(rootCause.getMessage(), rootCause);
239         doReturn(Futures.immediateFailedFuture(rootCause)).when(mockCohort2).canCommit();
240         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
241
242         FluentFuture<? extends CommitInfo> future = coordinator.commit(
243             transaction, Arrays.asList(mockCohort1, mockCohort2));
244
245         assertFailure(future, cause, mockCohort1, mockCohort2);
246     }
247
248     @Test
249     public void testSubmitWithPreCommitException() throws Exception {
250         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
251         doReturn(Futures.immediateFuture(null)).when(mockCohort1).preCommit();
252         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
253
254         doReturn(Futures.immediateFuture(true)).when(mockCohort2).canCommit();
255         IllegalStateException cause = new IllegalStateException("mock");
256         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).preCommit();
257         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
258
259         DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class);
260         doReturn(Futures.immediateFuture(true)).when(mockCohort3).canCommit();
261         doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock2")))
262                 .when(mockCohort3).preCommit();
263         doReturn(Futures.immediateFuture(null)).when(mockCohort3).abort();
264
265         FluentFuture<? extends CommitInfo> future = coordinator.commit(
266                 transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3));
267
268         assertFailure(future, cause, mockCohort1, mockCohort2, mockCohort3);
269     }
270
271     @Test
272     public void testSubmitWithCommitException() throws Exception {
273         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
274         doReturn(Futures.immediateFuture(null)).when(mockCohort1).preCommit();
275         doReturn(Futures.immediateFuture(null)).when(mockCohort1).commit();
276         doReturn(Futures.immediateFuture(null)).when(mockCohort1).abort();
277
278         doReturn(Futures.immediateFuture(true)).when(mockCohort2).canCommit();
279         doReturn(Futures.immediateFuture(null)).when(mockCohort2).preCommit();
280         IllegalStateException cause = new IllegalStateException("mock");
281         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).commit();
282         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
283
284         DOMStoreThreePhaseCommitCohort mockCohort3 = mock(DOMStoreThreePhaseCommitCohort.class);
285         doReturn(Futures.immediateFuture(true)).when(mockCohort3).canCommit();
286         doReturn(Futures.immediateFuture(null)).when(mockCohort3).preCommit();
287         doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock2")))
288                 .when(mockCohort3).commit();
289         doReturn(Futures.immediateFuture(null)).when(mockCohort3).abort();
290
291         FluentFuture<? extends CommitInfo> future = coordinator.commit(
292                 transaction, Arrays.asList(mockCohort1, mockCohort2, mockCohort3));
293
294         assertFailure(future, cause, mockCohort1, mockCohort2, mockCohort3);
295     }
296
297     @Test
298     public void testSubmitWithAbortException() throws Exception {
299         doReturn(Futures.immediateFuture(true)).when(mockCohort1).canCommit();
300         doReturn(Futures.immediateFailedFuture(new IllegalStateException("mock abort error")))
301                 .when(mockCohort1).abort();
302
303         IllegalStateException cause = new IllegalStateException("mock canCommit error");
304         doReturn(Futures.immediateFailedFuture(cause)).when(mockCohort2).canCommit();
305         doReturn(Futures.immediateFuture(null)).when(mockCohort2).abort();
306
307         FluentFuture<? extends CommitInfo> future = coordinator.commit(
308                 transaction, Arrays.asList(mockCohort1, mockCohort2));
309
310         assertFailure(future, cause, mockCohort1, mockCohort2);
311     }
312
313     @Test
314     public void testCreateReadWriteTransaction() {
315         DOMStore domStore = mock(DOMStore.class);
316         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
317                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
318                 futureExecutor)) {
319             dataBroker.newReadWriteTransaction();
320
321             verify(domStore, never()).newReadWriteTransaction();
322         }
323     }
324
325     @Test
326     public void testCreateWriteOnlyTransaction() {
327         DOMStore domStore = mock(DOMStore.class);
328         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
329                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
330                 futureExecutor)) {
331             dataBroker.newWriteOnlyTransaction();
332
333             verify(domStore, never()).newWriteOnlyTransaction();
334         }
335     }
336
337     @Test
338     public void testCreateReadOnlyTransaction() {
339         DOMStore domStore = mock(DOMStore.class);
340         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
341                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
342                 futureExecutor)) {
343             dataBroker.newReadOnlyTransaction();
344
345             verify(domStore, never()).newReadOnlyTransaction();
346         }
347     }
348
349     @Test
350     public void testLazySubTransactionCreationForReadWriteTransactions() {
351         DOMStore configDomStore = mock(DOMStore.class);
352         DOMStore operationalDomStore = mock(DOMStore.class);
353         DOMStoreReadWriteTransaction storeTxn = mock(DOMStoreReadWriteTransaction.class);
354
355         doReturn(storeTxn).when(operationalDomStore).newReadWriteTransaction();
356         doReturn(storeTxn).when(configDomStore).newReadWriteTransaction();
357
358         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
359                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
360                 configDomStore), futureExecutor)) {
361             DOMDataTreeReadWriteTransaction dataTxn = dataBroker.newReadWriteTransaction();
362
363             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
364             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
365             dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY);
366
367             verify(configDomStore, never()).newReadWriteTransaction();
368             verify(operationalDomStore, times(1)).newReadWriteTransaction();
369
370             dataTxn.put(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
371
372             verify(configDomStore, times(1)).newReadWriteTransaction();
373             verify(operationalDomStore, times(1)).newReadWriteTransaction();
374         }
375
376     }
377
378     @Test
379     public void testLazySubTransactionCreationForWriteOnlyTransactions() {
380         DOMStore configDomStore = mock(DOMStore.class);
381         DOMStore operationalDomStore = mock(DOMStore.class);
382         DOMStoreWriteTransaction storeTxn = mock(DOMStoreWriteTransaction.class);
383
384         doReturn(storeTxn).when(operationalDomStore).newWriteOnlyTransaction();
385         doReturn(storeTxn).when(configDomStore).newWriteOnlyTransaction();
386
387         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
388                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
389                 configDomStore), futureExecutor)) {
390             DOMDataTreeWriteTransaction dataTxn = dataBroker.newWriteOnlyTransaction();
391
392             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
393             dataTxn.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
394
395             verify(configDomStore, never()).newWriteOnlyTransaction();
396             verify(operationalDomStore, times(1)).newWriteOnlyTransaction();
397
398             dataTxn.put(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY, mock(NormalizedNode.class));
399
400             verify(configDomStore, times(1)).newWriteOnlyTransaction();
401             verify(operationalDomStore, times(1)).newWriteOnlyTransaction();
402         }
403     }
404
405     @Test
406     public void testLazySubTransactionCreationForReadOnlyTransactions() {
407         DOMStore configDomStore = mock(DOMStore.class);
408         DOMStore operationalDomStore = mock(DOMStore.class);
409         DOMStoreReadTransaction storeTxn = mock(DOMStoreReadTransaction.class);
410
411         doReturn(storeTxn).when(operationalDomStore).newReadOnlyTransaction();
412         doReturn(storeTxn).when(configDomStore).newReadOnlyTransaction();
413
414         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
415                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
416                 configDomStore), futureExecutor)) {
417             DOMDataTreeReadTransaction dataTxn = dataBroker.newReadOnlyTransaction();
418
419             dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY);
420             dataTxn.read(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY);
421
422             verify(configDomStore, never()).newReadOnlyTransaction();
423             verify(operationalDomStore, times(1)).newReadOnlyTransaction();
424
425             dataTxn.read(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY);
426
427             verify(configDomStore, times(1)).newReadOnlyTransaction();
428             verify(operationalDomStore, times(1)).newReadOnlyTransaction();
429         }
430     }
431
432     @Test
433     public void testSubmitWithOnlyOneSubTransaction() throws InterruptedException {
434         DOMStore configDomStore = mock(DOMStore.class);
435         DOMStore operationalDomStore = mock(DOMStore.class);
436         DOMStoreReadWriteTransaction mockStoreReadWriteTransaction = mock(DOMStoreReadWriteTransaction.class);
437         DOMStoreThreePhaseCommitCohort mockCohort = mock(DOMStoreThreePhaseCommitCohort.class);
438
439         doReturn(mockStoreReadWriteTransaction).when(operationalDomStore).newReadWriteTransaction();
440         doReturn(mockCohort).when(mockStoreReadWriteTransaction).ready();
441         doReturn(Futures.immediateFuture(false)).when(mockCohort).canCommit();
442         doReturn(Futures.immediateFuture(null)).when(mockCohort).abort();
443
444         final CountDownLatch latch = new CountDownLatch(1);
445         final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList<>();
446
447         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
448                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
449                 configDomStore), futureExecutor) {
450             @Override
451             public FluentFuture<? extends CommitInfo> commit(DOMDataTreeWriteTransaction writeTx,
452                     Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
453                 commitCohorts.addAll(cohorts);
454                 latch.countDown();
455                 return super.commit(writeTx, cohorts);
456             }
457         }) {
458             DOMDataTreeReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction();
459
460             domDataReadWriteTransaction.delete(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY);
461
462             domDataReadWriteTransaction.submit();
463
464             assertTrue(latch.await(10, TimeUnit.SECONDS));
465
466             assertTrue(commitCohorts.size() == 1);
467         }
468     }
469
470     @Test
471     public void testSubmitWithOnlyTwoSubTransactions() throws InterruptedException {
472         DOMStore configDomStore = mock(DOMStore.class);
473         DOMStore operationalDomStore = mock(DOMStore.class);
474         DOMStoreReadWriteTransaction operationalTransaction = mock(DOMStoreReadWriteTransaction.class);
475         DOMStoreReadWriteTransaction configTransaction = mock(DOMStoreReadWriteTransaction.class);
476         DOMStoreThreePhaseCommitCohort mockCohortOperational = mock(DOMStoreThreePhaseCommitCohort.class);
477         DOMStoreThreePhaseCommitCohort mockCohortConfig = mock(DOMStoreThreePhaseCommitCohort.class);
478
479         doReturn(operationalTransaction).when(operationalDomStore).newReadWriteTransaction();
480         doReturn(configTransaction).when(configDomStore).newReadWriteTransaction();
481
482         doReturn(mockCohortOperational).when(operationalTransaction).ready();
483         doReturn(Futures.immediateFuture(false)).when(mockCohortOperational).canCommit();
484         doReturn(Futures.immediateFuture(null)).when(mockCohortOperational).abort();
485
486         doReturn(mockCohortConfig).when(configTransaction).ready();
487         doReturn(Futures.immediateFuture(false)).when(mockCohortConfig).canCommit();
488         doReturn(Futures.immediateFuture(null)).when(mockCohortConfig).abort();
489
490         final CountDownLatch latch = new CountDownLatch(1);
491         final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList<>();
492
493         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
494                 LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION,
495                 configDomStore), futureExecutor) {
496             @Override
497             @SuppressWarnings("checkstyle:hiddenField")
498             public FluentFuture<? extends CommitInfo> commit(DOMDataTreeWriteTransaction writeTx,
499                     Collection<DOMStoreThreePhaseCommitCohort> cohorts) {
500                 commitCohorts.addAll(cohorts);
501                 latch.countDown();
502                 return super.commit(writeTx, cohorts);
503             }
504         }) {
505             DOMDataTreeReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction();
506
507             domDataReadWriteTransaction.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY,
508                     mock(NormalizedNode.class));
509             domDataReadWriteTransaction.merge(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.EMPTY,
510                     mock(NormalizedNode.class));
511
512             domDataReadWriteTransaction.submit();
513
514             assertTrue(latch.await(10, TimeUnit.SECONDS));
515
516             assertTrue(commitCohorts.size() == 2);
517         }
518     }
519
520     @Test
521     public void testCreateTransactionChain() {
522         DOMStore domStore = mock(DOMStore.class);
523         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
524                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
525                 futureExecutor)) {
526
527             dataBroker.createTransactionChain(mock(TransactionChainListener.class));
528
529             verify(domStore, times(2)).createTransactionChain();
530         }
531
532     }
533
534     @Test
535     public void testCreateTransactionOnChain() {
536         DOMStore domStore = mock(DOMStore.class);
537         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
538                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
539                 futureExecutor)) {
540
541             DOMStoreReadWriteTransaction operationalTransaction = mock(DOMStoreReadWriteTransaction.class);
542             DOMStoreTransactionChain mockChain = mock(DOMStoreTransactionChain.class);
543
544             doReturn(mockChain).when(domStore).createTransactionChain();
545             doReturn(operationalTransaction).when(mockChain).newWriteOnlyTransaction();
546
547             DOMTransactionChain transactionChain = dataBroker.createTransactionChain(
548                     mock(TransactionChainListener.class));
549
550             DOMDataTreeWriteTransaction domDataWriteTransaction = transactionChain.newWriteOnlyTransaction();
551
552             verify(mockChain, never()).newWriteOnlyTransaction();
553
554             domDataWriteTransaction.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.EMPTY,
555                     mock(NormalizedNode.class));
556         }
557     }
558
559     @Test
560     public void testEmptyTransactionSubmitSucceeds() throws ExecutionException, InterruptedException {
561         DOMStore domStore = mock(DOMStore.class);
562         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
563                 LogicalDatastoreType.OPERATIONAL, domStore, LogicalDatastoreType.CONFIGURATION, domStore),
564                 futureExecutor)) {
565
566             CheckedFuture<Void, TransactionCommitFailedException> submit1 =
567                     dataBroker.newWriteOnlyTransaction().submit();
568
569             assertNotNull(submit1);
570
571             submit1.get();
572
573             CheckedFuture<Void, TransactionCommitFailedException> submit2 =
574                     dataBroker.newReadWriteTransaction().submit();
575
576             assertNotNull(submit2);
577
578             submit2.get();
579         }
580     }
581
582     @Test
583     public void testExtensions() {
584         DistributedDataStore mockConfigStore = mock(DistributedDataStore.class);
585         DistributedDataStore mockOperStore = mock(DistributedDataStore.class);
586         try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of(
587                 LogicalDatastoreType.OPERATIONAL, mockOperStore,
588                 LogicalDatastoreType.CONFIGURATION, mockConfigStore), futureExecutor)) {
589
590             Map<Class<? extends DOMDataBrokerExtension>, DOMDataBrokerExtension> supportedExtensions =
591                     dataBroker.getSupportedExtensions();
592             assertNotNull(supportedExtensions.get(DOMDataTreeChangeService.class));
593
594             DOMDataTreeCommitCohortRegistry cohortRegistry =
595                     (DOMDataTreeCommitCohortRegistry) supportedExtensions.get(DOMDataTreeCommitCohortRegistry.class);
596             assertNotNull(cohortRegistry);
597
598             DOMDataTreeCommitCohort mockCohort = mock(DOMDataTreeCommitCohort.class);
599             DOMDataTreeIdentifier path = new DOMDataTreeIdentifier(
600                     org.opendaylight.mdsal.common.api.LogicalDatastoreType.CONFIGURATION,
601                     YangInstanceIdentifier.EMPTY);
602             cohortRegistry.registerCommitCohort(path, mockCohort);
603
604             verify(mockConfigStore).registerCommitCohort(path, mockCohort);
605         }
606     }
607 }

©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.