TransactionChainProxy extends AbstractTransactionContextFactory<LocalTransactionChain> implements DOMStoreTransactionChain { @Override public DOMStoreReadWriteTransaction newReadWriteTransaction() { getActorUtils().acquireTxCreationPermit(); return allocateWriteTransaction(TransactionType.READ_WRITE); } TransactionChainProxy(final TransactionContextFactory parent, final LocalHistoryIdentifier historyId); @Override DOMStoreReadTransaction newReadOnlyTransaction(); @Override DOMStoreReadWriteTransaction newReadWriteTransaction(); @Override DOMStoreWriteTransaction newWriteOnlyTransaction(); @Override void close();  }
@SuppressWarnings("resource") @Test public void testNewReadWriteTransaction() { DOMStoreTransaction dst = new TransactionChainProxy(mockComponentFactory, historyId).newReadWriteTransaction(); Assert.assertTrue(dst instanceof DOMStoreReadWriteTransaction); }
@Test public void testRateLimitingUsedInReadWriteTxCreation() { try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) { txChainProxy.newReadWriteTransaction(); verify(mockActorContext, times(1)).acquireTxCreationPermit(); } }
@Test @SuppressWarnings("checkstyle:IllegalCatch") public void testChainedReadWriteTransactions() throws Exception { try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) { ActorRef txActorRef1 = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE); expectBatchedModifications(txActorRef1, 1); Promise<Object> readyReplyPromise1 = akka.dispatch.Futures.promise(); doReturn(readyReplyPromise1.future()).when(mockActorContext).executeOperationAsync( eq(actorSelection(txActorRef1)), isA(BatchedModifications.class), any(Timeout.class)); DOMStoreWriteTransaction writeTx1 = txChainProxy.newReadWriteTransaction(); NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); writeTx1.write(TestModel.TEST_PATH, writeNode1); writeTx1.ready(); verifyOneBatchedModification(txActorRef1, new WriteModification(TestModel.TEST_PATH, writeNode1), true); String tx2MemberName = "mock-member"; ActorRef shardActorRef2 = setupActorContextWithoutInitialCreateTransaction(getSystem()); ActorRef txActorRef2 = setupActorContextWithInitialCreateTransaction(getSystem(), READ_WRITE, DataStoreVersions.CURRENT_VERSION, tx2MemberName, shardActorRef2); expectBatchedModifications(txActorRef2, 1); final NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME); final DOMStoreWriteTransaction writeTx2 = txChainProxy.newReadWriteTransaction(); final AtomicReference<Exception> caughtEx = new AtomicReference<>(); final CountDownLatch write2Complete = new CountDownLatch(1); new Thread(() -> { try { writeTx2.write(TestModel.OUTER_LIST_PATH, writeNode2); } catch (Exception e) { caughtEx.set(e); } finally { write2Complete.countDown(); } }).start(); assertTrue("Tx 2 write should've completed", write2Complete.await(5, TimeUnit.SECONDS)); if (caughtEx.get() != null) { throw caughtEx.get(); } try { verify(mockActorContext, never()).executeOperationAsync( eq(getSystem().actorSelection(shardActorRef2.path())), eqCreateTransaction(tx2MemberName, READ_WRITE)); } catch (AssertionError e) { fail("Tx 2 should not have initiated until the Tx 1's ready future completed"); } readyReplyPromise1.success(readyTxReply(txActorRef1.path().toString()).value().get().get()); verify(mockActorContext, timeout(5000)).executeOperationAsync( eq(getSystem().actorSelection(shardActorRef2.path())), eqCreateTransaction(tx2MemberName, READ_WRITE), any(Timeout.class)); } }