TransactionChainProxy extends AbstractTransactionContextFactory<LocalTransactionChain> implements DOMStoreTransactionChain { @Override public DOMStoreWriteTransaction newWriteOnlyTransaction() { getActorUtils().acquireTxCreationPermit(); return allocateWriteTransaction(TransactionType.WRITE_ONLY); } TransactionChainProxy(final TransactionContextFactory parent, final LocalHistoryIdentifier historyId); @Override DOMStoreReadTransaction newReadOnlyTransaction(); @Override DOMStoreReadWriteTransaction newReadWriteTransaction(); @Override DOMStoreWriteTransaction newWriteOnlyTransaction(); @Override void close();  }
@SuppressWarnings("resource") @Test public void testNewWriteOnlyTransaction() { DOMStoreTransaction dst = new TransactionChainProxy(mockComponentFactory, historyId).newWriteOnlyTransaction(); Assert.assertTrue(dst instanceof DOMStoreWriteTransaction); }
@Test public void testRateLimitingUsedInWriteOnlyTxCreation() { try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) { txChainProxy.newWriteOnlyTransaction(); verify(mockActorContext, times(1)).acquireTxCreationPermit(); } }
@Test @SuppressWarnings("checkstyle:IllegalCatch") public void testChainedWriteOnlyTransactions() throws Exception { dataStoreContextBuilder.writeOnlyTransactionOptimizationsEnabled(true); try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) { ActorRef txActorRef1 = setupActorContextWithoutInitialCreateTransaction(getSystem()); Promise<Object> batchedReplyPromise1 = akka.dispatch.Futures.promise(); doReturn(batchedReplyPromise1.future()).when(mockActorContext).executeOperationAsync( eq(actorSelection(txActorRef1)), isA(BatchedModifications.class), any(Timeout.class)); DOMStoreWriteTransaction writeTx1 = txChainProxy.newWriteOnlyTransaction(); NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); writeTx1.write(TestModel.TEST_PATH, writeNode1); writeTx1.ready(); verify(mockActorContext, times(1)).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); verifyOneBatchedModification(txActorRef1, new WriteModification(TestModel.TEST_PATH, writeNode1), true); ActorRef txActorRef2 = setupActorContextWithoutInitialCreateTransaction(getSystem()); expectBatchedModifications(txActorRef2, 1); final NormalizedNode<?, ?> writeNode2 = ImmutableNodes.containerNode(TestModel.OUTER_LIST_QNAME); final DOMStoreWriteTransaction writeTx2 = txChainProxy.newWriteOnlyTransaction(); 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, times(1)).findPrimaryShardAsync(eq(DefaultShardStrategy.DEFAULT_SHARD)); } catch (AssertionError e) { fail("Tx 2 should not have initiated until the Tx 1's ready future completed"); } batchedReplyPromise1.success(readyTxReply(txActorRef1.path().toString()).value().get().get()); verify(mockActorContext, timeout(5000).times(2)).findPrimaryShardAsync( eq(DefaultShardStrategy.DEFAULT_SHARD)); } }
@Test(expected = IllegalStateException.class) public void testChainedWriteTransactionsWithPreviousTxNotReady() { ActorRef actorRef = setupActorContextWithInitialCreateTransaction(getSystem(), WRITE_ONLY); expectBatchedModifications(actorRef, 1); try (TransactionChainProxy txChainProxy = new TransactionChainProxy(mockComponentFactory, historyId)) { DOMStoreWriteTransaction writeTx1 = txChainProxy.newWriteOnlyTransaction(); NormalizedNode<?, ?> writeNode1 = ImmutableNodes.containerNode(TestModel.TEST_QNAME); writeTx1.write(TestModel.TEST_PATH, writeNode1); txChainProxy.newWriteOnlyTransaction(); } }