plugins/netconf-client-mdsal/src/main/java/org/opendaylight/netconf/client/mdsal/spi/AbstractWriteTx.java

   1 /*
   2  * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.netconf.client.mdsal.spi;
   9
  10 import static com.google.common.base.Preconditions.checkArgument;
  11 import static com.google.common.base.Preconditions.checkState;
  12
  13 import com.google.common.util.concurrent.FluentFuture;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import com.google.common.util.concurrent.MoreExecutors;
  18 import com.google.common.util.concurrent.SettableFuture;
  19 import java.util.ArrayList;
  20 import java.util.List;
  21 import java.util.Optional;
  22 import java.util.concurrent.CopyOnWriteArrayList;
  23 import org.eclipse.jdt.annotation.NonNull;
  24 import org.opendaylight.mdsal.common.api.CommitInfo;
  25 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  26 import org.opendaylight.mdsal.common.api.TransactionCommitFailedException;
  27 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
  28 import org.opendaylight.mdsal.dom.api.DOMRpcResult;
  29 import org.opendaylight.netconf.api.EffectiveOperation;
  30 import org.opendaylight.netconf.api.NetconfDocumentedException;
  31 import org.opendaylight.netconf.client.mdsal.api.RemoteDeviceId;
  32 import org.opendaylight.netconf.client.mdsal.impl.NetconfBaseOps;
  33 import org.opendaylight.yangtools.concepts.AbstractRegistration;
  34 import org.opendaylight.yangtools.concepts.Registration;
  35 import org.opendaylight.yangtools.yang.common.ErrorSeverity;
  36 import org.opendaylight.yangtools.yang.common.ErrorTag;
  37 import org.opendaylight.yangtools.yang.common.ErrorType;
  38 import org.opendaylight.yangtools.yang.common.RpcError;
  39 import org.opendaylight.yangtools.yang.common.RpcResult;
  40 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  41 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  42 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
  43 import org.opendaylight.yangtools.yang.data.api.schema.MixinNode;
  44 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47
  48 abstract class AbstractWriteTx implements DOMDataTreeWriteTransaction {
  49     private static final Logger LOG  = LoggerFactory.getLogger(AbstractWriteTx.class);
  50
  51     final RemoteDeviceId id;
  52     final NetconfBaseOps netOps;
  53     final boolean rollbackSupport;
  54     final List<ListenableFuture<? extends DOMRpcResult>> resultsFutures = new ArrayList<>();
  55     private final List<TxListener> listeners = new CopyOnWriteArrayList<>();
  56     private final SettableFuture<CommitInfo> resultFuture = SettableFuture.create();
  57     private final @NonNull FluentFuture<CommitInfo> completionFuture = FluentFuture.from(resultFuture);
  58
  59     // Allow commit to be called only once
  60     volatile boolean finished = false;
  61     final boolean isLockAllowed;
  62
  63     AbstractWriteTx(final RemoteDeviceId id, final NetconfBaseOps netconfOps, final boolean rollbackSupport,
  64             final boolean isLockAllowed) {
  65         netOps = netconfOps;
  66         this.id = id;
  67         this.rollbackSupport = rollbackSupport;
  68         this.isLockAllowed = isLockAllowed;
  69     }
  70
  71     final void checkNotFinished() {
  72         checkState(!isFinished(), "%s: Transaction %s already finished", id, getIdentifier());
  73     }
  74
  75     final boolean isFinished() {
  76         return finished;
  77     }
  78
  79     abstract void init();
  80
  81     abstract void cleanup();
  82
  83     @Override
  84     public final FluentFuture<?> completionFuture() {
  85         return completionFuture;
  86     }
  87
  88     @Override
  89     public synchronized boolean cancel() {
  90         if (isFinished()) {
  91             return false;
  92         }
  93         listeners.forEach(listener -> listener.onTransactionCancelled(this));
  94         finished = true;
  95         cleanup();
  96         return true;
  97     }
  98
  99     @Override
 100     public final Object getIdentifier() {
 101         return this;
 102     }
 103
 104     @Override
 105     public final synchronized void put(final LogicalDatastoreType store, final YangInstanceIdentifier path,
 106             final NormalizedNode data) {
 107         checkEditable(store);
 108
 109         // Trying to write only mixin nodes (not visible when serialized).
 110         // Ignoring. Some devices cannot handle empty edit-config rpc
 111         if (containsOnlyNonVisibleData(path, data)) {
 112             LOG.debug("Ignoring put for {} and data {}. Resulting data structure is empty.", path, data);
 113             return;
 114         }
 115
 116         final var editStructure = netOps.createEditConfigStructure(Optional.ofNullable(data),
 117             Optional.of(EffectiveOperation.REPLACE), path);
 118         editConfig(path, Optional.ofNullable(data), editStructure, Optional.empty(), "put");
 119     }
 120
 121     @Override
 122     public final synchronized void merge(final LogicalDatastoreType store, final YangInstanceIdentifier path,
 123             final NormalizedNode data) {
 124         checkEditable(store);
 125
 126         // Trying to write only mixin nodes (not visible when serialized).
 127         // Ignoring. Some devices cannot handle empty edit-config rpc
 128         if (containsOnlyNonVisibleData(path, data)) {
 129             LOG.debug("Ignoring merge for {} and data {}. Resulting data structure is empty.", path, data);
 130             return;
 131         }
 132
 133         final var editStructure =  netOps.createEditConfigStructure(Optional.ofNullable(data), Optional.empty(), path);
 134         editConfig(path, Optional.ofNullable(data), editStructure, Optional.empty(), "merge");
 135     }
 136
 137     /**
 138      * Check whether the data to be written consists only from mixins.
 139      */
 140     private static boolean containsOnlyNonVisibleData(final YangInstanceIdentifier path, final NormalizedNode data) {
 141         // There's only one such case:top level list (pathArguments == 1 && data is Mixin)
 142         // any other mixin nodes are contained by a "regular" node thus visible when serialized
 143         return path.getPathArguments().size() == 1 && data instanceof MixinNode;
 144     }
 145
 146     @Override
 147     public final synchronized void delete(final LogicalDatastoreType store, final YangInstanceIdentifier path) {
 148         checkEditable(store);
 149         final DataContainerChild editStructure = netOps.createEditConfigStructure(Optional.empty(),
 150                         Optional.of(EffectiveOperation.DELETE), path);
 151         editConfig(path, Optional.empty(), editStructure, Optional.of(EffectiveOperation.NONE), "delete");
 152     }
 153
 154     @Override
 155     // Non-final for testing
 156     public FluentFuture<? extends CommitInfo> commit() {
 157         Futures.addCallback(commitConfiguration(), new FutureCallback<>() {
 158             @Override
 159             public void onSuccess(final RpcResult<Void> result) {
 160                 if (!result.isSuccessful()) {
 161                     resultFuture.setException(new TransactionCommitFailedException(
 162                         String.format("Commit of transaction %s failed", getIdentifier()),
 163                         result.getErrors().toArray(new RpcError[0])));
 164                     return;
 165                 }
 166
 167                 resultFuture.set(CommitInfo.empty());
 168             }
 169
 170             @Override
 171             public void onFailure(final Throwable failure) {
 172                 resultFuture.setException(new TransactionCommitFailedException(
 173                         String.format("Commit of transaction %s failed", getIdentifier()), failure));
 174             }
 175         }, MoreExecutors.directExecutor());
 176
 177         return completionFuture;
 178     }
 179
 180     final ListenableFuture<RpcResult<Void>> commitConfiguration() {
 181         listeners.forEach(listener -> listener.onTransactionSubmitted(this));
 182         checkNotFinished();
 183         finished = true;
 184         final var result = performCommit();
 185         Futures.addCallback(result, new FutureCallback<>() {
 186             @Override
 187             public void onSuccess(final RpcResult<Void> rpcResult) {
 188                 if (rpcResult.isSuccessful()) {
 189                     listeners.forEach(txListener -> txListener.onTransactionSuccessful(AbstractWriteTx.this));
 190                 } else {
 191                     final TransactionCommitFailedException cause =
 192                             new TransactionCommitFailedException("Transaction failed",
 193                                     rpcResult.getErrors().toArray(new RpcError[rpcResult.getErrors().size()]));
 194                     listeners.forEach(listener -> listener.onTransactionFailed(AbstractWriteTx.this, cause));
 195                 }
 196             }
 197
 198             @Override
 199             public void onFailure(final Throwable throwable) {
 200                 listeners.forEach(listener -> listener.onTransactionFailed(AbstractWriteTx.this, throwable));
 201             }
 202         }, MoreExecutors.directExecutor());
 203         return result;
 204     }
 205
 206     abstract ListenableFuture<RpcResult<Void>> performCommit();
 207
 208     private void checkEditable(final LogicalDatastoreType store) {
 209         checkNotFinished();
 210         checkArgument(store == LogicalDatastoreType.CONFIGURATION,
 211                 "Can edit only configuration data, not %s", store);
 212     }
 213
 214     abstract void editConfig(YangInstanceIdentifier path, Optional<NormalizedNode> data,
 215                              DataContainerChild editStructure, Optional<EffectiveOperation> defaultOperation,
 216                              String operation);
 217
 218     final ListenableFuture<RpcResult<Void>> resultsToTxStatus() {
 219         final var transformed = SettableFuture.<RpcResult<Void>>create();
 220
 221         Futures.addCallback(Futures.allAsList(resultsFutures), new FutureCallback<>() {
 222             @Override
 223             public void onSuccess(final List<DOMRpcResult> domRpcResults) {
 224                 if (!transformed.isDone()) {
 225                     extractResult(domRpcResults, transformed);
 226                 }
 227             }
 228
 229             @Override
 230             public void onFailure(final Throwable throwable) {
 231                 final NetconfDocumentedException exception =
 232                         new NetconfDocumentedException(
 233                                 id + ":RPC during tx returned an exception" + throwable.getMessage(),
 234                                 // FIXME: add proper unmask/wrap to ExecutionException
 235                                 new Exception(throwable),
 236                                 ErrorType.APPLICATION, ErrorTag.OPERATION_FAILED, ErrorSeverity.ERROR);
 237                 transformed.setException(exception);
 238             }
 239         }, MoreExecutors.directExecutor());
 240
 241         return transformed;
 242     }
 243
 244     private void extractResult(final List<DOMRpcResult> domRpcResults,
 245                                final SettableFuture<RpcResult<Void>> transformed) {
 246         ErrorType errType = ErrorType.APPLICATION;
 247         ErrorSeverity errSeverity = ErrorSeverity.ERROR;
 248         StringBuilder msgBuilder = new StringBuilder();
 249         boolean errorsEncouneterd = false;
 250         ErrorTag errorTag = ErrorTag.OPERATION_FAILED;
 251
 252         for (final DOMRpcResult domRpcResult : domRpcResults) {
 253             if (!domRpcResult.errors().isEmpty()) {
 254                 errorsEncouneterd = true;
 255                 final RpcError error = domRpcResult.errors().iterator().next();
 256
 257                 errType = error.getErrorType();
 258                 errSeverity = error.getSeverity();
 259                 msgBuilder.append(error.getMessage());
 260                 msgBuilder.append(error.getInfo());
 261                 errorTag = error.getTag();
 262             }
 263         }
 264         if (errorsEncouneterd) {
 265             final NetconfDocumentedException exception = new NetconfDocumentedException(
 266                     id + ":RPC during tx failed. " + msgBuilder, errType, errorTag, errSeverity);
 267             transformed.setException(exception);
 268             return;
 269         }
 270         transformed.set(RpcResultBuilder.<Void>success().build());
 271     }
 272
 273     // Non-final for testing
 274     Registration addListener(final TxListener listener) {
 275         listeners.add(listener);
 276         return new AbstractRegistration() {
 277             @Override
 278             protected void removeRegistration() {
 279                 listeners.remove(listener);
 280             }
 281         };
 282     }
 283 }