X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=netconf%2Fsal-netconf-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fnetconf%2Fsal%2Fconnect%2Fnetconf%2Fsal%2Ftx%2FWriteCandidateTx.java;h=1eb47c9775ba9651de88c56556b633f7f3132164;hb=7e1c9a3ba9211a98027bd9da7dc84a4b2136eb84;hp=b143a3bf8e969fb227e70f660862131c6eb509bf;hpb=b1b181f4cb695b36c4b1ecbd8f6371b31fbf40e4;p=netconf.git diff --git a/netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/tx/WriteCandidateTx.java b/netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/tx/WriteCandidateTx.java index b143a3bf8e..1eb47c9775 100644 --- a/netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/tx/WriteCandidateTx.java +++ b/netconf/sal-netconf-connector/src/main/java/org/opendaylight/netconf/sal/connect/netconf/sal/tx/WriteCandidateTx.java @@ -8,22 +8,18 @@ package org.opendaylight.netconf.sal.connect.netconf.sal.tx; -import com.google.common.base.Function; -import com.google.common.base.Optional; -import com.google.common.base.Preconditions; -import com.google.common.util.concurrent.CheckedFuture; +import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; -import org.opendaylight.controller.md.sal.common.api.TransactionStatus; -import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException; -import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult; -import org.opendaylight.netconf.api.NetconfDocumentedException; +import com.google.common.util.concurrent.MoreExecutors; +import java.util.Optional; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import org.opendaylight.mdsal.dom.api.DOMRpcResult; import org.opendaylight.netconf.sal.connect.netconf.util.NetconfBaseOps; import org.opendaylight.netconf.sal.connect.netconf.util.NetconfRpcFutureCallback; import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId; -import org.opendaylight.yangtools.yang.common.RpcError; import org.opendaylight.yangtools.yang.common.RpcResult; -import org.opendaylight.yangtools.yang.common.RpcResultBuilder; import org.opendaylight.yangtools.yang.data.api.ModifyAction; import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; import org.opendaylight.yangtools.yang.data.api.schema.DataContainerChild; @@ -32,7 +28,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * Tx implementation for netconf devices that support only candidate datastore and no writable running + * Tx implementation for netconf devices that support only candidate datastore and no writable running. * The sequence goes as: *
    *
  1. Lock candidate datastore on tx construction @@ -45,7 +41,8 @@ import org.slf4j.LoggerFactory; *
    2. *
  2. Edit-config in candidate N times * *
    3. *
  3. Commit and Unlock candidate datastore async
    4. @@ -55,22 +52,6 @@ public class WriteCandidateTx extends AbstractWriteTx { private static final Logger LOG = LoggerFactory.getLogger(WriteCandidateTx.class); - private static final Function> RPC_RESULT_TO_TX_STATUS = new Function>() { - @Override - public RpcResult apply(final DOMRpcResult input) { - if (isSuccess(input)) { - return RpcResultBuilder.success(TransactionStatus.COMMITED).build(); - } else { - final RpcResultBuilder failed = RpcResultBuilder.failed(); - for (final RpcError rpcError : input.getErrors()) { - failed.withError(rpcError.getErrorType(), rpcError.getTag(), rpcError.getMessage(), - rpcError.getApplicationTag(), rpcError.getInfo(), rpcError.getCause()); - } - return failed.build(); - } - } - }; - public WriteCandidateTx(final RemoteDeviceId id, final NetconfBaseOps rpc, final boolean rollbackSupport) { super(rpc, id, rollbackSupport); } @@ -78,34 +59,29 @@ public class WriteCandidateTx extends AbstractWriteTx { @Override protected synchronized void init() { LOG.trace("{}: Initializing {} transaction", id, getClass().getSimpleName()); - - try { - lock(); - } catch (final NetconfDocumentedException e) { - try { - LOG.warn("{}: Failed to lock candidate, attempting discard changes", id); - discardChanges(); - LOG.warn("{}: Changes discarded successfully, attempting lock", id); - lock(); - } catch (final NetconfDocumentedException secondE) { - LOG.error("{}: Failed to prepare candidate. Failed to initialize transaction", id, secondE); - throw new RuntimeException(id + ": Failed to prepare candidate. Failed to initialize transaction", secondE); - } - } + lock(); } - private void lock() throws NetconfDocumentedException { - try { - invokeBlocking("Lock candidate", new Function>() { - @Override - public ListenableFuture apply(final NetconfBaseOps input) { - return input.lockCandidate(new NetconfRpcFutureCallback("Lock candidate", id)); + private void lock() { + final FutureCallback lockCandidateCallback = new FutureCallback() { + @Override + public void onSuccess(@Nonnull final DOMRpcResult result) { + if (isSuccess(result)) { + if (LOG.isTraceEnabled()) { + LOG.trace("Lock candidate successful"); + } + } else { + LOG.warn("{}: lock candidate invoked unsuccessfully: {}", id, result.getErrors()); } - }); - } catch (final NetconfDocumentedException e) { - LOG.warn("{}: Failed to lock candidate", id, e); - throw e; - } + } + + @Override + public void onFailure(final Throwable throwable) { + LOG.warn("Lock candidate operation failed", throwable); + discardChanges(); + } + }; + resultsFutures.add(netOps.lockCandidate(lockCandidateCallback)); } @Override @@ -114,69 +90,34 @@ public class WriteCandidateTx extends AbstractWriteTx { cleanupOnSuccess(); } - @Override - protected void handleEditException(final YangInstanceIdentifier path, final NormalizedNode data, final NetconfDocumentedException e, final String editType) { - LOG.warn("{}: Error {} data to (candidate){}, data: {}, canceling", id, editType, path, data, e); - cancel(); - throw new RuntimeException(id + ": Error while " + editType + ": (candidate)" + path, e); - } - - @Override - protected void handleDeleteException(final YangInstanceIdentifier path, final NetconfDocumentedException e) { - LOG.warn("{}: Error deleting data (candidate){}, canceling", id, path, e); - cancel(); - throw new RuntimeException(id + ": Error while deleting (candidate)" + path, e); - } - - @Override - public synchronized CheckedFuture submit() { - final ListenableFuture commitFutureAsVoid = Futures.transform(commit(), new Function, Void>() { - @Override - public Void apply(final RpcResult input) { - Preconditions.checkArgument(input.isSuccessful() && input.getErrors().isEmpty(), "Submit failed with errors: %s", input.getErrors()); - return null; - } - }); - - return Futures.makeChecked(commitFutureAsVoid, new Function() { - @Override - public TransactionCommitFailedException apply(final Exception input) { - return new TransactionCommitFailedException("Submit of transaction " + getIdentifier() + " failed", input); - } - }); - } - /** - * 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 + * 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. */ private void discardChanges() { netOps.discardChanges(new NetconfRpcFutureCallback("Discarding candidate", id)); } @Override - public synchronized ListenableFuture> performCommit() { - final ListenableFuture rpcResult = netOps.commit(new NetconfRpcFutureCallback("Commit", id) { - @Override - public void onSuccess(final DOMRpcResult result) { - super.onSuccess(result); - LOG.debug("{}: Write successful, transaction: {}. Unlocking", id, getIdentifier()); - cleanupOnSuccess(); - } + public synchronized ListenableFuture> performCommit() { + resultsFutures.add(netOps.commit(new NetconfRpcFutureCallback("Commit", id))); + final ListenableFuture> txResult = resultsToTxStatus(); + Futures.addCallback(txResult, new FutureCallback>() { @Override - protected void onUnsuccess(final DOMRpcResult result) { - LOG.error("{}: Write failed, transaction {}, discarding changes, unlocking: {}", id, getIdentifier(), result.getErrors()); - cleanup(); + public void onSuccess(@Nullable final RpcResult result) { + cleanupOnSuccess(); } @Override - public void onFailure(final Throwable t) { - LOG.error("{}: Write failed, transaction {}, discarding changes, unlocking", id, getIdentifier(), t); + public void onFailure(final Throwable throwable) { + // TODO If lock is cause of this failure cleanup will issue warning log + // cleanup is trying to do unlock, but this will fail cleanup(); } - }); + }, MoreExecutors.directExecutor()); - return Futures.transform(rpcResult, RPC_RESULT_TO_TX_STATUS); + return txResult; } protected void cleanupOnSuccess() { @@ -184,21 +125,25 @@ public class WriteCandidateTx extends AbstractWriteTx { } @Override - protected void editConfig(final DataContainerChild editStructure, final Optional defaultOperation) throws NetconfDocumentedException { - invokeBlocking("Edit candidate", new Function>() { - @Override - public ListenableFuture apply(final NetconfBaseOps input) { - return defaultOperation.isPresent() - ? input.editConfigCandidate(new NetconfRpcFutureCallback("Edit candidate", id), editStructure, defaultOperation.get(), - rollbackSupport) - : input.editConfigCandidate(new NetconfRpcFutureCallback("Edit candidate", id), editStructure, - rollbackSupport); - } - }); + protected void editConfig(final YangInstanceIdentifier path, + final Optional> data, + final DataContainerChild editStructure, + final Optional defaultOperation, + final String operation) { + + final NetconfRpcFutureCallback editConfigCallback = new NetconfRpcFutureCallback("Edit candidate", id); + + if (defaultOperation.isPresent()) { + resultsFutures.add(netOps.editConfigCandidate( + editConfigCallback, editStructure, defaultOperation.get(), rollbackSupport)); + } else { + resultsFutures.add(netOps.editConfigCandidate(editConfigCallback, editStructure, rollbackSupport)); + } } /** - * 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 + * 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. */ private void unlock() { netOps.unlockCandidate(new NetconfRpcFutureCallback("Unlock candidate", id));