netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/tx/WriteCandidateTx.java

   1 /*
   2  * Copyright (c) 2014 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
   9 package org.opendaylight.netconf.sal.connect.netconf.sal.tx;
  10
  11 import com.google.common.base.Function;
  12 import com.google.common.base.Optional;
  13 import com.google.common.base.Preconditions;
  14 import com.google.common.util.concurrent.CheckedFuture;
  15 import com.google.common.util.concurrent.FutureCallback;
  16 import com.google.common.util.concurrent.Futures;
  17 import com.google.common.util.concurrent.ListenableFuture;
  18 import javax.annotation.Nullable;
  19 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
  20 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  21 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
  22 import org.opendaylight.netconf.sal.connect.netconf.util.NetconfBaseOps;
  23 import org.opendaylight.netconf.sal.connect.netconf.util.NetconfRpcFutureCallback;
  24 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
  25 import org.opendaylight.yangtools.yang.common.RpcError;
  26 import org.opendaylight.yangtools.yang.common.RpcResult;
  27 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  28 import org.opendaylight.yangtools.yang.data.api.ModifyAction;
  29 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  30 import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild;
  31 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34
  35 /**
  36  * Tx implementation for netconf devices that support only candidate datastore and no writable running
  37  * The sequence goes as:
  38  * <ol>
  39  *   <li>Lock candidate datastore on tx construction
  40  *     <ul>
  41  *       <li>Lock has to succeed, if it does not, an attempt to discard changes is made</li>
  42  *       <li>Discard changes has to succeed</li>
  43  *       <li>If discard is successful, lock is reattempted</li>
  44  *       <li>Second lock attempt has to succeed</li>
  45  *     </ul>
  46  *   </li>
  47  *   <li>Edit-config in candidate N times
  48  *     <ul>
  49  *       <li>If any issue occurs during edit, datastore is discarded using discard-changes rpc, unlocked and an exception is thrown async</li>
  50  *     </ul>
  51  *   </li>
  52  *   <li>Commit and Unlock candidate datastore async</li>
  53  * </ol>
  54  */
  55 public class WriteCandidateTx extends AbstractWriteTx {
  56
  57     private static final Logger LOG  = LoggerFactory.getLogger(WriteCandidateTx.class);
  58
  59     private static final Function<DOMRpcResult, RpcResult<TransactionStatus>> RPC_RESULT_TO_TX_STATUS = new Function<DOMRpcResult, RpcResult<TransactionStatus>>() {
  60         @Override
  61         public RpcResult<TransactionStatus> apply(final DOMRpcResult input) {
  62             if (isSuccess(input)) {
  63                 return RpcResultBuilder.success(TransactionStatus.COMMITED).build();
  64             } else {
  65                 final RpcResultBuilder<TransactionStatus> failed = RpcResultBuilder.failed();
  66                 for (final RpcError rpcError : input.getErrors()) {
  67                     failed.withError(rpcError.getErrorType(), rpcError.getTag(), rpcError.getMessage(),
  68                             rpcError.getApplicationTag(), rpcError.getInfo(), rpcError.getCause());
  69                 }
  70                 return failed.build();
  71             }
  72         }
  73     };
  74
  75     public WriteCandidateTx(final RemoteDeviceId id, final NetconfBaseOps rpc, final boolean rollbackSupport) {
  76         super(rpc, id, rollbackSupport);
  77     }
  78
  79     @Override
  80     protected synchronized void init() {
  81         LOG.trace("{}: Initializing {} transaction", id, getClass().getSimpleName());
  82         lock();
  83     }
  84
  85     private void lock() {
  86         final FutureCallback<DOMRpcResult> lockCandidateCallback = new FutureCallback<DOMRpcResult>() {
  87             @Override
  88             public void onSuccess(DOMRpcResult result) {
  89                 if (isSuccess(result)) {
  90                     if (LOG.isTraceEnabled()) {
  91                         LOG.trace("Lock candidate successful");
  92                     }
  93                 } else {
  94                     LOG.warn("{}: lock candidate invoked unsuccessfully: {}", id, result.getErrors());
  95                 }
  96             }
  97
  98             @Override
  99             public void onFailure(Throwable t) {
 100                 LOG.warn("Lock candidate operation failed. {}", t);
 101                 discardChanges();
 102             }
 103         };
 104         resultsFutures.add(netOps.lockCandidate(lockCandidateCallback));
 105     }
 106
 107     @Override
 108     protected void cleanup() {
 109         discardChanges();
 110         cleanupOnSuccess();
 111     }
 112
 113     @Override
 114     public synchronized CheckedFuture<Void, TransactionCommitFailedException> submit() {
 115         final ListenableFuture<Void> commitFutureAsVoid = Futures.transform(commit(), new Function<RpcResult<TransactionStatus>, Void>() {
 116             @Override
 117             public Void apply(final RpcResult<TransactionStatus> input) {
 118                 Preconditions.checkArgument(input.isSuccessful() && input.getErrors().isEmpty(), "Submit failed with errors: %s", input.getErrors());
 119                 return null;
 120             }
 121         });
 122
 123         return Futures.makeChecked(commitFutureAsVoid, new Function<Exception, TransactionCommitFailedException>() {
 124             @Override
 125             public TransactionCommitFailedException apply(final Exception input) {
 126                 return new TransactionCommitFailedException("Submit of transaction " + getIdentifier() + " failed", input);
 127             }
 128         });
 129     }
 130
 131     /**
 132      * This has to be non blocking since it is called from a callback on commit and its netty threadpool that is really sensitive to blocking calls
 133      */
 134     private void discardChanges() {
 135         netOps.discardChanges(new NetconfRpcFutureCallback("Discarding candidate", id));
 136     }
 137
 138     @Override
 139     public synchronized ListenableFuture<RpcResult<TransactionStatus>> performCommit() {
 140         resultsFutures.add(netOps.commit(new NetconfRpcFutureCallback("Commit", id)));
 141         ListenableFuture<RpcResult<TransactionStatus>> txResult = resultsToTxStatus();
 142
 143         Futures.addCallback(txResult, new FutureCallback<RpcResult<TransactionStatus>>() {
 144             @Override
 145             public void onSuccess(@Nullable RpcResult<TransactionStatus> result) {
 146                 cleanupOnSuccess();
 147             }
 148
 149             @Override
 150             public void onFailure(Throwable t) {
 151                 // TODO If lock is cause of this failure cleanup will issue warning log
 152                 // cleanup is trying to do unlock, but this will fail
 153                 cleanup();
 154             }
 155         });
 156
 157         return txResult;
 158     }
 159
 160     protected void cleanupOnSuccess() {
 161         unlock();
 162     }
 163
 164     @Override
 165     protected void editConfig(final YangInstanceIdentifier path,
 166                               final Optional<NormalizedNode<?, ?>> data,
 167                               final DataContainerChild<?, ?> editStructure,
 168                               final Optional<ModifyAction> defaultOperation,
 169                               final String operation) {
 170
 171         NetconfRpcFutureCallback editConfigCallback = new NetconfRpcFutureCallback("Edit candidate", id);
 172
 173         if (defaultOperation.isPresent()) {
 174             resultsFutures.add(netOps.editConfigCandidate(
 175                     editConfigCallback, editStructure, defaultOperation.get(), rollbackSupport));
 176         } else {
 177             resultsFutures.add(netOps.editConfigCandidate(editConfigCallback, editStructure, rollbackSupport));
 178         }
 179     }
 180
 181     /**
 182      * This has to be non blocking since it is called from a callback on commit and its netty threadpool that is really sensitive to blocking calls
 183      */
 184     private void unlock() {
 185         netOps.unlockCandidate(new NetconfRpcFutureCallback("Unlock candidate", id));
 186     }
 187
 188 }