+ @Test
+ public void testPipelinedTransactionsWithImmediateCommits() throws Exception {
+ final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
+ snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
+
+ final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
+ snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
+
+ YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
+ MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
+ final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
+
+ final FutureCallback<UnsignedLong> commitCallback2 = immediate3PhaseCommit(cohort2);
+ final FutureCallback<UnsignedLong> commitCallback3 = immediate3PhaseCommit(cohort3);
+ final FutureCallback<UnsignedLong> commitCallback1 = immediate3PhaseCommit(cohort1);
+
+ InOrder inOrder = inOrder(mockShard);
+ inOrder.verify(mockShard).persistPayload(eq(cohort1.getIdentifier()), any(CommitTransactionPayload.class),
+ eq(true));
+ inOrder.verify(mockShard).persistPayload(eq(cohort2.getIdentifier()), any(CommitTransactionPayload.class),
+ eq(true));
+ inOrder.verify(mockShard).persistPayload(eq(cohort3.getIdentifier()), any(CommitTransactionPayload.class),
+ eq(false));
+
+ // The payload instance doesn't matter - it just needs to be of type CommitTransactionPayload.
+ CommitTransactionPayload mockPayload = CommitTransactionPayload.create(nextTransactionId(),
+ cohort1.getCandidate());
+ shardDataTree.applyReplicatedPayload(cohort1.getIdentifier(), mockPayload);
+ shardDataTree.applyReplicatedPayload(cohort2.getIdentifier(), mockPayload);
+ shardDataTree.applyReplicatedPayload(cohort3.getIdentifier(), mockPayload);
+
+ inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3);
+ inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
+ inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
+ inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
+
+ final DataTreeSnapshot snapshot =
+ shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
+ Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(carPath);
+ assertTrue("Car node present", optional.isPresent());
+ assertEquals("Car node", carNode, optional.get());
+ }
+
+ @Test
+ public void testPipelinedTransactionsWithImmediateReplication() {
+ immediatePayloadReplication(shardDataTree, mockShard);
+
+ final ShardDataTreeCohort cohort1 = newShardDataTreeCohort(snapshot ->
+ snapshot.write(CarsModel.BASE_PATH, CarsModel.emptyContainer()));
+
+ final ShardDataTreeCohort cohort2 = newShardDataTreeCohort(snapshot ->
+ snapshot.write(CarsModel.CAR_LIST_PATH, CarsModel.newCarMapNode()));
+
+ YangInstanceIdentifier carPath = CarsModel.newCarPath("optima");
+ MapEntryNode carNode = CarsModel.newCarEntry("optima", new BigInteger("100"));
+ final ShardDataTreeCohort cohort3 = newShardDataTreeCohort(snapshot -> snapshot.write(carPath, carNode));
+
+ final FutureCallback<UnsignedLong> commitCallback1 = immediate3PhaseCommit(cohort1);
+ final FutureCallback<UnsignedLong> commitCallback2 = immediate3PhaseCommit(cohort2);
+ final FutureCallback<UnsignedLong> commitCallback3 = immediate3PhaseCommit(cohort3);
+
+ InOrder inOrder = inOrder(commitCallback1, commitCallback2, commitCallback3);
+ inOrder.verify(commitCallback1).onSuccess(any(UnsignedLong.class));
+ inOrder.verify(commitCallback2).onSuccess(any(UnsignedLong.class));
+ inOrder.verify(commitCallback3).onSuccess(any(UnsignedLong.class));
+
+ final DataTreeSnapshot snapshot = shardDataTree.newReadOnlyTransaction(nextTransactionId()).getSnapshot();
+ Optional<NormalizedNode<?, ?>> optional = snapshot.readNode(CarsModel.BASE_PATH);
+ assertTrue("Car node present", optional.isPresent());
+ }
+