@Test public void testSubmitWithOnlyTwoSubTransactions() throws InterruptedException { DOMStore configDomStore = mock(DOMStore.class); DOMStore operationalDomStore = mock(DOMStore.class); DOMStoreReadWriteTransaction operationalTransaction = mock(DOMStoreReadWriteTransaction.class); DOMStoreReadWriteTransaction configTransaction = mock(DOMStoreReadWriteTransaction.class); DOMStoreThreePhaseCommitCohort mockCohortOperational = mock(DOMStoreThreePhaseCommitCohort.class); DOMStoreThreePhaseCommitCohort mockCohortConfig = mock(DOMStoreThreePhaseCommitCohort.class); doReturn(operationalTransaction).when(operationalDomStore).newReadWriteTransaction(); doReturn(configTransaction).when(configDomStore).newReadWriteTransaction(); doReturn(mockCohortOperational).when(operationalTransaction).ready(); doReturn(immediateFalseFluentFuture()).when(mockCohortOperational).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohortOperational).abort(); doReturn(mockCohortConfig).when(configTransaction).ready(); doReturn(immediateFalseFluentFuture()).when(mockCohortConfig).canCommit(); doReturn(immediateNullFluentFuture()).when(mockCohortConfig).abort(); final CountDownLatch latch = new CountDownLatch(1); final List<DOMStoreThreePhaseCommitCohort> commitCohorts = new ArrayList<>(); try (ConcurrentDOMDataBroker dataBroker = new ConcurrentDOMDataBroker(ImmutableMap.of( LogicalDatastoreType.OPERATIONAL, operationalDomStore, LogicalDatastoreType.CONFIGURATION, configDomStore), futureExecutor) { @Override @SuppressWarnings("checkstyle:hiddenField") public FluentFuture<? extends CommitInfo> commit(DOMDataTreeWriteTransaction writeTx, Collection<DOMStoreThreePhaseCommitCohort> cohorts) { commitCohorts.addAll(cohorts); latch.countDown(); return super.commit(writeTx, cohorts); } }) { DOMDataTreeReadWriteTransaction domDataReadWriteTransaction = dataBroker.newReadWriteTransaction(); domDataReadWriteTransaction.put(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.empty(), mock(NormalizedNode.class)); domDataReadWriteTransaction.merge(LogicalDatastoreType.CONFIGURATION, YangInstanceIdentifier.empty(), mock(NormalizedNode.class)); domDataReadWriteTransaction.commit(); assertTrue(latch.await(10, TimeUnit.SECONDS)); assertTrue(commitCohorts.size() == 2); } }
@Override protected FluentFuture<? extends CommitInfo> commit(final DOMDataTreeWriteTransaction transaction, final Collection<DOMStoreThreePhaseCommitCohort> cohorts) { checkArgument(transaction != null, "Transaction must not be null."); checkArgument(cohorts != null, "Cohorts must not be null."); LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier()); if (cohorts.isEmpty()) { return CommitInfo.emptyFluentFuture(); } final AsyncNotifyingSettableFuture clientSubmitFuture = new AsyncNotifyingSettableFuture(clientFutureCallbackExecutor); doCanCommit(clientSubmitFuture, transaction, cohorts); return FluentFuture.from(clientSubmitFuture).transform(ignored -> CommitInfo.empty(), MoreExecutors.directExecutor()); }
ConcurrentDOMDataBroker extends AbstractDOMBroker { @Override protected FluentFuture<? extends CommitInfo> commit(final DOMDataTreeWriteTransaction transaction, final Collection<DOMStoreThreePhaseCommitCohort> cohorts) { checkArgument(transaction != null, "Transaction must not be null."); checkArgument(cohorts != null, "Cohorts must not be null."); LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier()); if (cohorts.isEmpty()) { return CommitInfo.emptyFluentFuture(); } final AsyncNotifyingSettableFuture clientSubmitFuture = new AsyncNotifyingSettableFuture(clientFutureCallbackExecutor); doCanCommit(clientSubmitFuture, transaction, cohorts); return FluentFuture.from(clientSubmitFuture).transform(ignored -> CommitInfo.empty(), MoreExecutors.directExecutor()); } }
ConcurrentDOMDataBroker extends AbstractDOMBroker { @Override protected FluentFuture<? extends CommitInfo> commit(final DOMDataTreeWriteTransaction transaction, final Collection<DOMStoreThreePhaseCommitCohort> cohorts) { checkArgument(transaction != null, "Transaction must not be null."); checkArgument(cohorts != null, "Cohorts must not be null."); LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier()); if (cohorts.isEmpty()) { return CommitInfo.emptyFluentFuture(); } final AsyncNotifyingSettableFuture clientSubmitFuture = new AsyncNotifyingSettableFuture(clientFutureCallbackExecutor); doCanCommit(clientSubmitFuture, transaction, cohorts); return FluentFuture.from(clientSubmitFuture).transform(ignored -> CommitInfo.empty(), MoreExecutors.directExecutor()); } ConcurrentDOMDataBroker(final Map<LogicalDatastoreType, DOMStore> datastores,
            final Executor listenableFutureExecutor);  ConcurrentDOMDataBroker(final Map<LogicalDatastoreType, DOMStore> datastores,
            final Executor listenableFutureExecutor, final DurationStatisticsTracker commitStatsTracker); }
ConcurrentDOMDataBroker extends AbstractDOMBroker { @Override protected FluentFuture<? extends CommitInfo> commit(final DOMDataTreeWriteTransaction transaction, final Collection<DOMStoreThreePhaseCommitCohort> cohorts) { checkArgument(transaction != null, "Transaction must not be null."); checkArgument(cohorts != null, "Cohorts must not be null."); LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier()); if (cohorts.isEmpty()) { return CommitInfo.emptyFluentFuture(); } final AsyncNotifyingSettableFuture clientSubmitFuture = new AsyncNotifyingSettableFuture(clientFutureCallbackExecutor); doCanCommit(clientSubmitFuture, transaction, cohorts); return FluentFuture.from(clientSubmitFuture).transform(ignored -> CommitInfo.empty(), MoreExecutors.directExecutor()); } ConcurrentDOMDataBroker(final Map<LogicalDatastoreType, DOMStore> datastores,
            final Executor listenableFutureExecutor);  ConcurrentDOMDataBroker(final Map<LogicalDatastoreType, DOMStore> datastores,
            final Executor listenableFutureExecutor, final DurationStatisticsTracker commitStatsTracker); DurationStatisticsTracker getCommitStatsTracker(); @Override String toString(); }
ConcurrentDOMDataBroker extends AbstractDOMBroker { @Override protected FluentFuture<? extends CommitInfo> commit(final DOMDataTreeWriteTransaction transaction, final Collection<DOMStoreThreePhaseCommitCohort> cohorts) { checkArgument(transaction != null, "Transaction must not be null."); checkArgument(cohorts != null, "Cohorts must not be null."); LOG.debug("Tx: {} is submitted for execution.", transaction.getIdentifier()); if (cohorts.isEmpty()) { return CommitInfo.emptyFluentFuture(); } final AsyncNotifyingSettableFuture clientSubmitFuture = new AsyncNotifyingSettableFuture(clientFutureCallbackExecutor); doCanCommit(clientSubmitFuture, transaction, cohorts); return FluentFuture.from(clientSubmitFuture).transform(ignored -> CommitInfo.empty(), MoreExecutors.directExecutor()); } ConcurrentDOMDataBroker(final Map<LogicalDatastoreType, DOMStore> datastores,
            final Executor listenableFutureExecutor);  ConcurrentDOMDataBroker(final Map<LogicalDatastoreType, DOMStore> datastores,
            final Executor listenableFutureExecutor, final DurationStatisticsTracker commitStatsTracker); DurationStatisticsTracker getCommitStatsTracker(); @Override String toString();  }