import org.opendaylight.mdsal.dom.spi.store.DOMStore;
import org.opendaylight.mdsal.dom.spi.store.DOMStoreThreePhaseCommitCohort;
import org.opendaylight.yangtools.util.DurationStatisticsTracker;
+import org.opendaylight.yangtools.yang.common.Empty;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public ConcurrentDOMDataBroker(final Map<LogicalDatastoreType, DOMStore> datastores,
final Executor listenableFutureExecutor, final DurationStatisticsTracker commitStatsTracker) {
super(datastores);
- this.clientFutureCallbackExecutor = requireNonNull(listenableFutureExecutor);
+ clientFutureCallbackExecutor = requireNonNull(listenableFutureExecutor);
this.commitStatsTracker = requireNonNull(commitStatsTracker);
}
doCanCommit(clientSubmitFuture, transaction, cohorts);
- return FluentFuture.from(clientSubmitFuture).transform(ignored -> CommitInfo.empty(),
- MoreExecutors.directExecutor());
+ return FluentFuture.from(clientSubmitFuture);
}
private void doCanCommit(final AsyncNotifyingSettableFuture clientSubmitFuture,
final Iterator<DOMStoreThreePhaseCommitCohort> cohortIterator = cohorts.iterator();
// Not using Futures.allAsList here to avoid its internal overhead.
- FutureCallback<Boolean> futureCallback = new FutureCallback<Boolean>() {
+ FutureCallback<Boolean> futureCallback = new FutureCallback<>() {
@Override
public void onSuccess(final Boolean result) {
if (result == null || !result) {
}
};
- ListenableFuture<Boolean> canCommitFuture = cohortIterator.next().canCommit();
- Futures.addCallback(canCommitFuture, futureCallback, MoreExecutors.directExecutor());
+ Futures.addCallback(cohortIterator.next().canCommit(), futureCallback, MoreExecutors.directExecutor());
}
private void doPreCommit(final long startTime, final AsyncNotifyingSettableFuture clientSubmitFuture,
final Iterator<DOMStoreThreePhaseCommitCohort> cohortIterator = cohorts.iterator();
// Not using Futures.allAsList here to avoid its internal overhead.
- FutureCallback<Void> futureCallback = new FutureCallback<Void>() {
+ FutureCallback<Empty> futureCallback = new FutureCallback<>() {
@Override
- public void onSuccess(final Void notUsed) {
+ public void onSuccess(final Empty result) {
if (!cohortIterator.hasNext()) {
// All cohorts completed successfully - we can move on to the commit phase
doCommit(startTime, clientSubmitFuture, transaction, cohorts);
} else {
- ListenableFuture<Void> preCommitFuture = cohortIterator.next().preCommit();
- Futures.addCallback(preCommitFuture, this, MoreExecutors.directExecutor());
+ Futures.addCallback(cohortIterator.next().preCommit(), this, MoreExecutors.directExecutor());
}
}
}
};
- ListenableFuture<Void> preCommitFuture = cohortIterator.next().preCommit();
- Futures.addCallback(preCommitFuture, futureCallback, MoreExecutors.directExecutor());
+ Futures.addCallback(cohortIterator.next().preCommit(), futureCallback, MoreExecutors.directExecutor());
}
private void doCommit(final long startTime, final AsyncNotifyingSettableFuture clientSubmitFuture,
final Iterator<DOMStoreThreePhaseCommitCohort> cohortIterator = cohorts.iterator();
// Not using Futures.allAsList here to avoid its internal overhead.
- FutureCallback<Void> futureCallback = new FutureCallback<Void>() {
+ final FutureCallback<CommitInfo> futureCallback = new FutureCallback<>() {
@Override
- public void onSuccess(final Void notUsed) {
+ public void onSuccess(final CommitInfo result) {
if (!cohortIterator.hasNext()) {
// All cohorts completed successfully - we're done.
commitStatsTracker.addDuration(System.nanoTime() - startTime);
clientSubmitFuture.set();
} else {
- ListenableFuture<Void> commitFuture = cohortIterator.next().commit();
- Futures.addCallback(commitFuture, this, MoreExecutors.directExecutor());
+ Futures.addCallback(cohortIterator.next().commit(), this, MoreExecutors.directExecutor());
}
}
}
};
- ListenableFuture<Void> commitFuture = cohortIterator.next().commit();
- Futures.addCallback(commitFuture, futureCallback, MoreExecutors.directExecutor());
+ Futures.addCallback(cohortIterator.next().commit(), futureCallback, MoreExecutors.directExecutor());
}
@SuppressFBWarnings(value = "BC_UNCONFIRMED_CAST_OF_RETURN_VALUE",
// Transaction failed - tell all cohorts to abort.
@SuppressWarnings("unchecked")
- ListenableFuture<Void>[] canCommitFutures = new ListenableFuture[cohorts.size()];
+ ListenableFuture<Empty>[] canCommitFutures = new ListenableFuture[cohorts.size()];
int index = 0;
for (DOMStoreThreePhaseCommitCohort cohort : cohorts) {
canCommitFutures[index++] = cohort.abort();
}
clientSubmitFuture.setException(exMapper.apply(e));
- ListenableFuture<List<Void>> combinedFuture = Futures.allAsList(canCommitFutures);
- Futures.addCallback(combinedFuture, new FutureCallback<List<Void>>() {
+ ListenableFuture<List<Empty>> combinedFuture = Futures.allAsList(canCommitFutures);
+ Futures.addCallback(combinedFuture, new FutureCallback<List<Empty>>() {
@Override
- public void onSuccess(final List<Void> notUsed) {
+ public void onSuccess(final List<Empty> result) {
// Propagate the original exception to the client.
LOG.debug("Tx: {} aborted successfully", transaction.getIdentifier());
}
* FIXME: This class should probably be moved to yangtools common utils for re-usability and
* unified with AsyncNotifyingListenableFutureTask.
*/
- private static class AsyncNotifyingSettableFuture extends AbstractFuture<Void> {
-
+ private static class AsyncNotifyingSettableFuture extends AbstractFuture<CommitInfo> {
/**
* ThreadLocal used to detect if the task completion thread is running the future listener Runnables.
*/
boolean set() {
ON_TASK_COMPLETION_THREAD_TL.set(Boolean.TRUE);
try {
- return super.set(null);
+ return super.set(CommitInfo.empty());
} finally {
ON_TASK_COMPLETION_THREAD_TL.set(null);
}