aa38e30fe84f587bd20f8fcbd26e38b8a4f9fa1c
[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / test / java / org / opendaylight / controller / cluster / datastore / ShardDataTreeTest.java
1 /*
2  * Copyright (c) 2015 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 static org.junit.Assert.assertEquals;
12 import static org.junit.Assert.assertNotNull;
13 import static org.junit.Assert.fail;
14 import static org.mockito.Matchers.any;
15 import static org.mockito.Matchers.anyBoolean;
16 import static org.mockito.Matchers.eq;
17 import static org.mockito.Mockito.atLeastOnce;
18 import static org.mockito.Mockito.doNothing;
19 import static org.mockito.Mockito.doReturn;
20 import static org.mockito.Mockito.inOrder;
21 import static org.mockito.Mockito.mock;
22 import static org.mockito.Mockito.never;
23 import static org.mockito.Mockito.reset;
24 import static org.mockito.Mockito.verify;
25 import static org.mockito.Mockito.verifyNoMoreInteractions;
26 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.coordinatedCanCommit;
27 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.coordinatedCommit;
28 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.coordinatedPreCommit;
29 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediate3PhaseCommit;
30 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediateCanCommit;
31 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediateCommit;
32 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediatePayloadReplication;
33 import static org.opendaylight.controller.cluster.datastore.ShardDataTreeMocking.immediatePreCommit;
34
35 import com.google.common.base.Ticker;
36 import com.google.common.collect.Maps;
37 import com.google.common.primitives.UnsignedLong;
38 import com.google.common.util.concurrent.FutureCallback;
39 import java.math.BigInteger;
40 import java.util.ArrayList;
41 import java.util.Collection;
42 import java.util.List;
43 import java.util.Map;
44 import java.util.Optional;
45 import java.util.concurrent.ExecutionException;
46 import java.util.function.Consumer;
47 import org.junit.Before;
48 import org.junit.Test;
49 import org.mockito.ArgumentCaptor;
50 import org.mockito.InOrder;
51 import org.mockito.Mockito;
52 import org.opendaylight.controller.cluster.datastore.jmx.mbeans.shard.ShardStats;
53 import org.opendaylight.controller.cluster.datastore.persisted.CommitTransactionPayload;
54 import org.opendaylight.controller.md.cluster.datastore.model.CarsModel;
55 import org.opendaylight.controller.md.cluster.datastore.model.PeopleModel;
56 import org.opendaylight.controller.md.cluster.datastore.model.SchemaContextHelper;
57 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
58 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
59 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
60 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
61 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
62 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
63 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeModification;
64 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
65 import org.opendaylight.yangtools.yang.data.api.schema.tree.ModificationType;
66 import org.opendaylight.yangtools.yang.data.api.schema.tree.TreeType;
67 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
68
69 public class ShardDataTreeTest extends AbstractTest {
70
71     private final Shard mockShard = Mockito.mock(Shard.class);
72     private ShardDataTree shardDataTree;
73     private SchemaContext fullSchema;
74
75     @Before
76     public void setUp() {
77         doReturn(Ticker.systemTicker()).when(mockShard).ticker();
78         doReturn(Mockito.mock(ShardStats.class)).when(mockShard).getShardMBean();
79
80         fullSchema = SchemaContextHelper.full();
81
82         shardDataTree = new ShardDataTree(mockShard, fullSchema, TreeType.OPERATIONAL);
83     }
84
85     @Test
86     public void testWrite() throws ExecutionException, InterruptedException {
87         modify(false, true, true);
88     }
89
90     @Test
91     public void testMerge() throws ExecutionException, InterruptedException {
92         modify(true, true, true);
93     }
94
95     private void modify(final boolean merge, final boolean expectedCarsPresent, final boolean expectedPeoplePresent)
96             throws ExecutionException, InterruptedException {
97         immediatePayloadReplication(shardDataTree, mockShard);
98
99         assertEquals(fullSchema, shardDataTree.getSchemaContext());
100
101         final ReadWriteShardDataTreeTransaction transaction =
102                 shardDataTree.newReadWriteTransaction(nextTransactionId());
103
104         final DataTreeModification snapshot = transaction.getSnapshot();
105
106         assertNotNull(snapshot);
107
108         if (merge) {
109             snapshot.merge(CarsModel.BASE_PATH, CarsModel.create());
110             snapshot.merge(PeopleModel.BASE_PATH, PeopleModel.create());
111         } else {
112             snapshot.write(CarsModel.BASE_PATH, CarsModel.create());
113             snapshot.write(PeopleModel.BASE_PATH, PeopleModel.create());
114         }
115
116         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction);
117
118         immediateCanCommit(cohort);
119         immediatePreCommit(cohort);
120         immediateCommit(cohort);
121
122         final ReadOnlyShardDataTreeTransaction readOnlyShardDataTreeTransaction =
123                 shardDataTree.newReadOnlyTransaction(nextTransactionId());
124
125         final DataTreeSnapshot snapshot1 = readOnlyShardDataTreeTransaction.getSnapshot();
126
127         final Optional<NormalizedNode<?, ?>> optional = snapshot1.readNode(CarsModel.BASE_PATH);
128
129         assertEquals(expectedCarsPresent, optional.isPresent());
130
131         final Optional<NormalizedNode<?, ?>> optional1 = snapshot1.readNode(PeopleModel.BASE_PATH);
132
133         assertEquals(expectedPeoplePresent, optional1.isPresent());
134     }
135
136     @Test
137     public void bug4359AddRemoveCarOnce() throws ExecutionException, InterruptedException {
138         immediatePayloadReplication(shardDataTree, mockShard);
139
140         final List<DataTreeCandidate> candidates = new ArrayList<>();
141         candidates.add(addCar(shardDataTree));
142         candidates.add(removeCar(shardDataTree));
143
144         final NormalizedNode<?, ?> expected = getCars(shardDataTree);
145
146         applyCandidates(shardDataTree, candidates);
147
148         final NormalizedNode<?, ?> actual = getCars(shardDataTree);
149
150         assertEquals(expected, actual);
151     }
152
153     @Test
154     public void bug4359AddRemoveCarTwice() throws ExecutionException, InterruptedException {
155         immediatePayloadReplication(shardDataTree, mockShard);
156
157         final List<DataTreeCandidate> candidates = new ArrayList<>();
158         candidates.add(addCar(shardDataTree));
159         candidates.add(removeCar(shardDataTree));
160         candidates.add(addCar(shardDataTree));
161         candidates.add(removeCar(shardDataTree));
162
163         final NormalizedNode<?, ?> expected = getCars(shardDataTree);
164
165         applyCandidates(shardDataTree, candidates);
166
167         final NormalizedNode<?, ?> actual = getCars(shardDataTree);
168
169         assertEquals(expected, actual);
170     }
171
172     @Test
173     public void testListenerNotifiedOnApplySnapshot() throws Exception {
174         immediatePayloadReplication(shardDataTree, mockShard);
175
176         DOMDataTreeChangeListener listener = mock(DOMDataTreeChangeListener.class);
177         shardDataTree.registerTreeChangeListener(CarsModel.CAR_LIST_PATH.node(CarsModel.CAR_QNAME), listener,
178             com.google.common.base.Optional.absent(), noop -> { });
179
180         addCar(shardDataTree, "optima");
181
182         verifyOnDataTreeChanged(listener, dtc -> {
183             assertEquals("getModificationType", ModificationType.WRITE, dtc.getRootNode().getModificationType());
184             assertEquals("getRootPath", CarsModel.newCarPath("optima"), dtc.getRootPath());
185         });
186
187         addCar(shardDataTree, "sportage");
188
189         verifyOnDataTreeChanged(listener, dtc -> {
190             assertEquals("getModificationType", ModificationType.WRITE, dtc.getRootNode().getModificationType());
191             assertEquals("getRootPath", CarsModel.newCarPath("sportage"), dtc.getRootPath());
192         });
193
194         ShardDataTree newDataTree = new ShardDataTree(mockShard, fullSchema, TreeType.OPERATIONAL);
195         immediatePayloadReplication(newDataTree, mockShard);
196         addCar(newDataTree, "optima");
197         addCar(newDataTree, "murano");
198
199         shardDataTree.applySnapshot(newDataTree.takeStateSnapshot());
200
201         Map<YangInstanceIdentifier, ModificationType> expChanges = Maps.newHashMap();
202         expChanges.put(CarsModel.newCarPath("optima"), ModificationType.WRITE);
203         expChanges.put(CarsModel.newCarPath("murano"), ModificationType.WRITE);
204         expChanges.put(CarsModel.newCarPath("sportage"), ModificationType.DELETE);
205         verifyOnDataTreeChanged(listener, dtc -> {
206             ModificationType expType = expChanges.remove(dtc.getRootPath());
207             assertNotNull("Got unexpected change for " + dtc.getRootPath(), expType);
208             assertEquals("getModificationType", expType, dtc.getRootNode().getModificationType());
209         });
210
211         if (!expChanges.isEmpty()) {
212             fail("Missing change notifications: " + expChanges);
213         }
214     }
215
216     @Test
217     public void testPipelinedTransactionsWithCoordinatedCommits() throws Exception {
218         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
219             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
220
221         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
222             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
223
224         NormalizedNode<?, ?> peopleNode = PeopleModel.create();
225         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
226             snapshot.write(PeopleModel.BASE_PATH, peopleNode));
227
228         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
229         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
230         final ShardDataTreeCohort cohort4 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
231
232         immediateCanCommit(cohort1);
233         final FutureCallback<Void> canCommitCallback2 = coordinatedCanCommit(cohort2);
234         final FutureCallback<Void> canCommitCallback3 = coordinatedCanCommit(cohort3);
235         final FutureCallback<Void> canCommitCallback4 = coordinatedCanCommit(cohort4);
236
237         final FutureCallback<DataTreeCandidate> preCommitCallback1 = coordinatedPreCommit(cohort1);
238         verify(preCommitCallback1).onSuccess(cohort1.getCandidate());
239         verify(canCommitCallback2).onSuccess(null);
240
241         final FutureCallback<DataTreeCandidate> preCommitCallback2 = coordinatedPreCommit(cohort2);
242         verify(preCommitCallback2).onSuccess(cohort2.getCandidate());
243         verify(canCommitCallback3).onSuccess(null);
244
245         final FutureCallback<DataTreeCandidate> preCommitCallback3 = coordinatedPreCommit(cohort3);
246         verify(preCommitCallback3).onSuccess(cohort3.getCandidate());
247         verify(canCommitCallback4).onSuccess(null);
248
249         final FutureCallback<DataTreeCandidate> preCommitCallback4 = coordinatedPreCommit(cohort4);
250         verify(preCommitCallback4).onSuccess(cohort4.getCandidate());
251
252         final FutureCallback<UnsignedLong> commitCallback2 = coordinatedCommit(cohort2);
253         verify(mockShard, never()).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
254                 anyBoolean());
255         verifyNoMoreInteractions(commitCallback2);
256
257         final FutureCallback<UnsignedLong> commitCallback4 = coordinatedCommit(cohort4);
258         verify(mockShard, never()).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
259                 anyBoolean());
260         verifyNoMoreInteractions(commitCallback4);
261
262         final FutureCallback<UnsignedLong> commitCallback1 = coordinatedCommit(cohort1);
263         InOrder inOrder = inOrder(mockShard);
264         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
265                 eq(true));
266         inOrder.verify(mockShard).persistPayload(eq(cohort2.getIdentifier()), any(CommitTransactionPayload.class),
267                 eq(false));
268         verifyNoMoreInteractions(commitCallback1);
269         verifyNoMoreInteractions(commitCallback2);
270
271         final FutureCallback<UnsignedLong> commitCallback3 = coordinatedCommit(cohort3);
272         inOrder = inOrder(mockShard);
273         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
274                 eq(true));
275         inOrder.verify(mockShard).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
276                 eq(false));
277         verifyNoMoreInteractions(commitCallback3);
278         verifyNoMoreInteractions(commitCallback4);
279
280         final ShardDataTreeCohort cohort5 = newShardDataTreeCohort(snapshot ->
281             snapshot.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
282         final FutureCallback<Void> canCommitCallback5 = coordinatedCanCommit(cohort5);
283
284         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
285         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
286                 cohort1.getCandidate());
287         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
288         shardDataTree.applyReplicatedPayload(cohort2.getIdentifier(), mockPayload);
289         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
290         shardDataTree.applyReplicatedPayload(cohort4.getIdentifier(), mockPayload);
291
292         inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3, commitCallback4);
293         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
294         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
295         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
296         inOrder.verify(commitCallback4).onSuccess(any(UnsignedLong.class));
297
298         verify(canCommitCallback5).onSuccess(null);
299
300         final DataTreeSnapshot snapshot =
301                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
302         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
303         assertEquals("Car node present", true, optional.isPresent());
304         assertEquals("Car node", carNode, optional.get());
305
306         optional = snapshot.readNode(PeopleModel.BASE_PATH);
307         assertEquals("People node present", true, optional.isPresent());
308         assertEquals("People node", peopleNode, optional.get());
309     }
310
311     @Test
312     public void testPipelinedTransactionsWithImmediateCommits() throws Exception {
313         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
314             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
315
316         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
317             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
318
319         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
320         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
321         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
322
323         final FutureCallback<UnsignedLong> commitCallback2 = immediate3PhaseCommit(cohort2);
324         final FutureCallback<UnsignedLong> commitCallback3 = immediate3PhaseCommit(cohort3);
325         final FutureCallback<UnsignedLong> commitCallback1 = immediate3PhaseCommit(cohort1);
326
327         InOrder inOrder = inOrder(mockShard);
328         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
329                 eq(true));
330         inOrder.verify(mockShard).persistPayload(eq(cohort2.getIdentifier()), any(CommitTransactionPayload.class),
331                 eq(true));
332         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
333                 eq(false));
334
335         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
336         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
337                 cohort1.getCandidate());
338         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
339         shardDataTree.applyReplicatedPayload(cohort2.getIdentifier(), mockPayload);
340         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
341
342         inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3);
343         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
344         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
345         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
346
347         final DataTreeSnapshot snapshot =
348                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
349         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
350         assertEquals("Car node present", true, optional.isPresent());
351         assertEquals("Car node", carNode, optional.get());
352     }
353
354     @Test
355     public void testPipelinedTransactionsWithImmediateReplication() throws Exception {
356         immediatePayloadReplication(shardDataTree, mockShard);
357
358         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
359             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
360
361         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
362             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
363
364         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
365         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
366         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
367
368         final FutureCallback<UnsignedLong> commitCallback1 = immediate3PhaseCommit(cohort1);
369         final FutureCallback<UnsignedLong> commitCallback2 = immediate3PhaseCommit(cohort2);
370         final FutureCallback<UnsignedLong> commitCallback3 = immediate3PhaseCommit(cohort3);
371
372         InOrder inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3);
373         inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
374         inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
375         inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
376
377         final DataTreeSnapshot snapshot =
378                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
379         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(CarsModel.BASE_PATH);
380         assertEquals("Car node present", true, optional.isPresent());
381     }
382
383     @SuppressWarnings("unchecked")
384     @Test
385     public void testAbortWithPendingCommits() throws Exception {
386         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
387             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
388
389         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
390             snapshot.write(PeopleModel.BASE_PATH, PeopleModel.create()));
391
392         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
393             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
394
395         YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
396         MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
397         final ShardDataTreeCohort cohort4 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
398
399         coordinatedCanCommit(cohort2);
400         immediateCanCommit(cohort1);
401         coordinatedCanCommit(cohort3);
402         coordinatedCanCommit(cohort4);
403
404         coordinatedPreCommit(cohort1);
405         coordinatedPreCommit(cohort2);
406         coordinatedPreCommit(cohort3);
407
408         FutureCallback<Void> mockAbortCallback = mock(FutureCallback.class);
409         doNothing().when(mockAbortCallback).onSuccess(null);
410         cohort2.abort(mockAbortCallback);
411         verify(mockAbortCallback).onSuccess(null);
412
413         coordinatedPreCommit(cohort4);
414         coordinatedCommit(cohort1);
415         coordinatedCommit(cohort3);
416         coordinatedCommit(cohort4);
417
418         InOrder inOrder = inOrder(mockShard);
419         inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
420                 eq(false));
421         inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
422                 eq(false));
423         inOrder.verify(mockShard).persistPayload(eq(cohort4.getIdentifier()), any(CommitTransactionPayload.class),
424                 eq(false));
425
426         // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
427         CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
428                 cohort1.getCandidate());
429         shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
430         shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
431         shardDataTree.applyReplicatedPayload(cohort4.getIdentifier(), mockPayload);
432
433         final DataTreeSnapshot snapshot =
434                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
435         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
436         assertEquals("Car node present", true, optional.isPresent());
437         assertEquals("Car node", carNode, optional.get());
438     }
439
440     @SuppressWarnings("unchecked")
441     @Test
442     public void testAbortWithFailedRebase() throws Exception {
443         immediatePayloadReplication(shardDataTree, mockShard);
444
445         final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
446             snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
447
448         final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
449             snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
450
451         NormalizedNode<?, ?> peopleNode = PeopleModel.create();
452         final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot ->
453             snapshot.write(PeopleModel.BASE_PATH, peopleNode));
454
455         immediateCanCommit(cohort1);
456         FutureCallback<Void> canCommitCallback2 = coordinatedCanCommit(cohort2);
457
458         coordinatedPreCommit(cohort1);
459         verify(canCommitCallback2).onSuccess(null);
460
461         FutureCallback<Void> mockAbortCallback = mock(FutureCallback.class);
462         doNothing().when(mockAbortCallback).onSuccess(null);
463         cohort1.abort(mockAbortCallback);
464         verify(mockAbortCallback).onSuccess(null);
465
466         FutureCallback<DataTreeCandidate> preCommitCallback2 = coordinatedPreCommit(cohort2);
467         verify(preCommitCallback2).onFailure(any(Throwable.class));
468
469         immediateCanCommit(cohort3);
470         immediatePreCommit(cohort3);
471         immediateCommit(cohort3);
472
473         final DataTreeSnapshot snapshot =
474                 shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
475         Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(PeopleModel.BASE_PATH);
476         assertEquals("People node present", true, optional.isPresent());
477         assertEquals("People node", peopleNode, optional.get());
478     }
479
480     private ShardDataTreeCohort newShardDataTreeCohort(final DataTreeOperation operation) {
481         final ReadWriteShardDataTreeTransaction transaction =
482                 shardDataTree.newReadWriteTransaction(nextTransactionId());
483         final DataTreeModification snapshot = transaction.getSnapshot();
484         operation.execute(snapshot);
485         return shardDataTree.finishTransaction(transaction);
486     }
487
488     @SuppressWarnings({ "rawtypes", "unchecked" })
489     private static void verifyOnDataTreeChanged(final DOMDataTreeChangeListener listener,
490             final Consumer<DataTreeCandidate> callback) {
491         ArgumentCaptor<Collection> changes = ArgumentCaptor.forClass(Collection.class);
492         verify(listener, atLeastOnce()).onDataTreeChanged(changes.capture());
493         for (Collection list : changes.getAllValues()) {
494             for (Object dtc : list) {
495                 callback.accept((DataTreeCandidate)dtc);
496             }
497         }
498
499         reset(listener);
500     }
501
502     private static NormalizedNode<?, ?> getCars(final ShardDataTree shardDataTree) {
503         final ReadOnlyShardDataTreeTransaction readOnlyShardDataTreeTransaction =
504                 shardDataTree.newReadOnlyTransaction(nextTransactionId());
505         final DataTreeSnapshot snapshot1 = readOnlyShardDataTreeTransaction.getSnapshot();
506
507         final Optional<NormalizedNode<?, ?>> optional = snapshot1.readNode(CarsModel.BASE_PATH);
508
509         assertEquals(true, optional.isPresent());
510
511         return optional.get();
512     }
513
514     private static DataTreeCandidate addCar(final ShardDataTree shardDataTree)
515             throws ExecutionException, InterruptedException {
516         return addCar(shardDataTree, "altima");
517     }
518
519     private static DataTreeCandidate addCar(final ShardDataTree shardDataTree, final String name)
520             throws ExecutionException, InterruptedException {
521         return doTransaction(shardDataTree, snapshot -> {
522             snapshot.merge(CarsModel.BASE_PATH, CarsModel.emptyContainer());
523             snapshot.merge(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode());
524             snapshot.write(CarsModel.newCarPath(name), CarsModel.newCarEntry(name, new BigInteger("100")));
525         });
526     }
527
528     private static DataTreeCandidate removeCar(final ShardDataTree shardDataTree)
529             throws ExecutionException, InterruptedException {
530         return doTransaction(shardDataTree, snapshot -> snapshot.delete(CarsModel.newCarPath("altima")));
531     }
532
533     @FunctionalInterface
534     private interface DataTreeOperation {
535         void execute(DataTreeModification snapshot);
536     }
537
538     private static DataTreeCandidate doTransaction(final ShardDataTree shardDataTree,
539             final DataTreeOperation operation) throws ExecutionException, InterruptedException {
540         final ReadWriteShardDataTreeTransaction transaction =
541                 shardDataTree.newReadWriteTransaction(nextTransactionId());
542         final DataTreeModification snapshot = transaction.getSnapshot();
543         operation.execute(snapshot);
544         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction);
545
546         immediateCanCommit(cohort);
547         immediatePreCommit(cohort);
548         final DataTreeCandidate candidate = cohort.getCandidate();
549         immediateCommit(cohort);
550
551         return candidate;
552     }
553
554     private static DataTreeCandidate applyCandidates(final ShardDataTree shardDataTree,
555             final List<DataTreeCandidate> candidates) throws ExecutionException, InterruptedException {
556         final ReadWriteShardDataTreeTransaction transaction =
557                 shardDataTree.newReadWriteTransaction(nextTransactionId());
558         final DataTreeModification snapshot = transaction.getSnapshot();
559         for (final DataTreeCandidate candidateTip : candidates) {
560             DataTreeCandidates.applyToModification(snapshot, candidateTip);
561         }
562         final ShardDataTreeCohort cohort = shardDataTree.finishTransaction(transaction);
563
564         immediateCanCommit(cohort);
565         immediatePreCommit(cohort);
566         final DataTreeCandidate candidate = cohort.getCandidate();
567         immediateCommit(cohort);
568
569         return candidate;
570     }
571 }

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