+ final SettableFuture<CommitInfo> commitResult = SettableFuture.create();
+
+ // First complete all resultsFutures and merge them ...
+ final ListenableFuture<DOMRpcResult> resultErrors = mergeFutures(resultsFutures);
+
+ // ... then evaluate if there are any problems
+ Futures.addCallback(resultErrors, new FutureCallback<>() {
+ @Override
+ public void onSuccess(final DOMRpcResult result) {
+ final Collection<? extends RpcError> errors = result.getErrors();
+ if (!allWarnings(errors)) {
+ Futures.whenAllComplete(discardAndUnlock()).run(
+ () -> commitResult.setException(toCommitFailedException(errors)),
+ MoreExecutors.directExecutor());
+ return;
+ }
+
+ // ... no problems so far, initiate commit
+ Futures.addCallback(netconfService.commit(), new FutureCallback<DOMRpcResult>() {
+ @Override
+ public void onSuccess(final DOMRpcResult rpcResult) {
+ final Collection<? extends RpcError> errors = result.getErrors();
+ if (errors.isEmpty()) {
+ Futures.whenAllComplete(netconfService.unlock()).run(
+ () -> commitResult.set(CommitInfo.empty()),
+ MoreExecutors.directExecutor());
+ } else if (allWarnings(errors)) {
+ LOG.info("Commit successful with warnings {}", errors);
+ Futures.whenAllComplete(netconfService.unlock()).run(
+ () -> commitResult.set(CommitInfo.empty()),
+ MoreExecutors.directExecutor());
+ } else {
+ Futures.whenAllComplete(discardAndUnlock()).run(
+ () -> commitResult.setException(toCommitFailedException(errors)),
+ MoreExecutors.directExecutor());
+ }
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ Futures.whenAllComplete(discardAndUnlock()).run(
+ () -> commitResult.setException(throwable),
+ MoreExecutors.directExecutor());
+ }
+ }, MoreExecutors.directExecutor());
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ Futures.whenAllComplete(discardAndUnlock()).run(() -> commitResult.setException(throwable),
+ MoreExecutors.directExecutor());
+ }
+ }, MoreExecutors.directExecutor());
+
+ return FluentFuture.from(commitResult);
+ }
+
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
+ private List<ListenableFuture<?>> discardAndUnlock() {
+ // execute discard & unlock operations only if lock operation was completed successfully
+ if (isLocked) {
+ return List.of(netconfService.discardChanges(), netconfService.unlock());
+ } else {
+ return Collections.emptyList();
+ }
+ }
+
+ private final FutureCallback<DOMRpcResult> lockOperationCallback = new FutureCallback<>() {
+ @Override
+ public void onSuccess(final DOMRpcResult rpcResult) {
+ if (rpcResult != null && allWarnings(rpcResult.getErrors())) {
+ isLocked = true;
+ }
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ // do nothing
+ }
+ };
+
+ private void enqueueOperation(final Supplier<ListenableFuture<? extends DOMRpcResult>> operation) {
+ final ListenableFuture<? extends DOMRpcResult> operationFuture;
+ synchronized (resultsFutures) {
+ // if we only have result for the lock operation ...
+ if (resultsFutures.size() == 1) {
+ operationFuture = Futures.transformAsync(resultsFutures.get(0),
+ result -> {
+ // ... then add new operation to the chain if lock was successful
+ if (result != null && (result.getErrors().isEmpty() || allWarnings(result.getErrors()))) {
+ return operation.get();
+ } else {
+ return Futures.immediateFailedFuture(new NetconfDocumentedException("Lock operation failed",
+ DocumentedException.ErrorType.APPLICATION, DocumentedException.ErrorTag.LOCK_DENIED,
+ ErrorSeverity.ERROR));
+ }
+ },
+ MoreExecutors.directExecutor());
+ } else {
+ // ... otherwise just add operation to the execution chain
+ operationFuture = Futures.transformAsync(resultsFutures.get(resultsFutures.size() - 1),
+ future -> operation.get(),
+ MoreExecutors.directExecutor());
+ }
+ // ... finally save operation related future to the list
+ resultsFutures.add(operationFuture);
+ }
+ }
+
+ // Transform list of futures related to RPC operation into a single Future
+ private static ListenableFuture<DOMRpcResult> mergeFutures(
+ final List<ListenableFuture<? extends DOMRpcResult>> futures) {
+ return Futures.whenAllComplete(futures).call(() -> {
+ if (futures.size() == 1) {
+ // Fast path
+ return Futures.getDone(futures.get(0));
+ }
+
+ final var builder = ImmutableList.<RpcError>builder();
+ for (ListenableFuture<? extends DOMRpcResult> future : futures) {
+ builder.addAll(Futures.getDone(future).getErrors());
+ }
+ return new DefaultDOMRpcResult(null, builder.build());
+ }, MoreExecutors.directExecutor());
+ }
+
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
+ private static TransactionCommitFailedException toCommitFailedException(
+ final Collection<? extends RpcError> errors) {
+ DocumentedException.ErrorType errType = DocumentedException.ErrorType.APPLICATION;
+ ErrorSeverity errSeverity = ErrorSeverity.ERROR;
+ StringJoiner msgBuilder = new StringJoiner(" ");
+ String errorTag = "operation-failed";
+ for (final RpcError error : errors) {
+ switch (error.getErrorType()) {
+ case RPC:
+ errType = DocumentedException.ErrorType.RPC;
+ break;
+ case PROTOCOL:
+ errType = DocumentedException.ErrorType.PROTOCOL;
+ break;
+ case TRANSPORT:
+ errType = DocumentedException.ErrorType.TRANSPORT;
+ break;
+ case APPLICATION:
+ default:
+ errType = DocumentedException.ErrorType.APPLICATION;
+ break;
+ }
+ errSeverity = error.getSeverity().toNetconf();
+ msgBuilder.add(error.getMessage());
+ msgBuilder.add(error.getInfo());
+ errorTag = error.getTag();
+ }
+
+ return new TransactionCommitFailedException("Netconf transaction commit failed",
+ new NetconfDocumentedException("RPC during tx failed. " + msgBuilder.toString(), errType,
+ DocumentedException.ErrorTag.from(errorTag), errSeverity));
+ }
+
+ private static void executeWithLogging(final Supplier<ListenableFuture<? extends DOMRpcResult>> operation) {
+ final ListenableFuture<? extends DOMRpcResult> operationResult = operation.get();
+ Futures.addCallback(operationResult, new FutureCallback<DOMRpcResult>() {
+ @Override
+ public void onSuccess(final DOMRpcResult rpcResult) {
+ if (rpcResult != null && !rpcResult.getErrors().isEmpty()) {
+ LOG.error("Errors occurred during processing of the RPC operation: {}",
+ rpcResult.getErrors().stream().map(Object::toString).collect(Collectors.joining(",")));
+ }
+ }
+
+ @Override
+ public void onFailure(final Throwable throwable) {
+ LOG.error("Error processing operation", throwable);
+ }
+ }, MoreExecutors.directExecutor());
+ }
+
+ @SuppressFBWarnings(value = "UPM_UNCALLED_PRIVATE_METHOD",
+ justification = "https://github.com/spotbugs/spotbugs/issues/811")
+ private static boolean allWarnings(final Collection<? extends @NonNull RpcError> errors) {
+ return errors.stream().allMatch(error -> error.getSeverity() == RpcError.ErrorSeverity.WARNING);