2 * Copyright (c) 2017 Pantheon Technologies s.r.o. and others. All rights reserved.
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
9 package org.opendaylight.netconf.topology.singleton.impl.tx;
11 import akka.actor.ActorRef;
12 import akka.actor.ActorSystem;
13 import akka.dispatch.OnComplete;
14 import akka.pattern.Patterns;
15 import akka.util.Timeout;
16 import com.google.common.base.Preconditions;
17 import com.google.common.util.concurrent.FluentFuture;
18 import com.google.common.util.concurrent.SettableFuture;
19 import java.util.concurrent.atomic.AtomicBoolean;
20 import org.eclipse.jdt.annotation.NonNull;
21 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
22 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
23 import org.opendaylight.mdsal.common.api.CommitInfo;
24 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
25 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
26 import org.opendaylight.netconf.topology.singleton.messages.NormalizedNodeMessage;
27 import org.opendaylight.netconf.topology.singleton.messages.transactions.CancelRequest;
28 import org.opendaylight.netconf.topology.singleton.messages.transactions.DeleteRequest;
29 import org.opendaylight.netconf.topology.singleton.messages.transactions.MergeRequest;
30 import org.opendaylight.netconf.topology.singleton.messages.transactions.PutRequest;
31 import org.opendaylight.netconf.topology.singleton.messages.transactions.SubmitFailedReply;
32 import org.opendaylight.netconf.topology.singleton.messages.transactions.SubmitRequest;
33 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
34 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
35 import org.slf4j.Logger;
36 import org.slf4j.LoggerFactory;
37 import scala.concurrent.Await;
38 import scala.concurrent.Future;
40 public class ProxyWriteAdapter {
42 private static final Logger LOG = LoggerFactory.getLogger(ProxyWriteAdapter.class);
44 private final ActorRef masterTxActor;
45 private final RemoteDeviceId id;
46 private final ActorSystem actorSystem;
47 private final AtomicBoolean opened = new AtomicBoolean(true);
48 private final Timeout askTimeout;
50 public ProxyWriteAdapter(final ActorRef masterTxActor, final RemoteDeviceId id, final ActorSystem actorSystem,
51 final Timeout askTimeout) {
52 this.masterTxActor = masterTxActor;
54 this.actorSystem = actorSystem;
55 this.askTimeout = askTimeout;
58 @SuppressWarnings("checkstyle:IllegalCatch")
59 public boolean cancel() {
60 if (!opened.compareAndSet(true, false)) {
63 final Future<Object> cancelScalaFuture =
64 Patterns.ask(masterTxActor, new CancelRequest(), askTimeout);
66 LOG.trace("{}: Cancel {} via NETCONF", id);
69 // here must be Await because AsyncWriteTransaction do not return future
70 return (boolean) Await.result(cancelScalaFuture, askTimeout.duration());
71 } catch (final Exception e) {
76 public @NonNull FluentFuture<? extends @NonNull CommitInfo> commit(final Object identifier) {
77 if (!opened.compareAndSet(true, false)) {
78 throw new IllegalStateException(id + ": Transaction" + identifier + " is closed");
80 final Future<Object> submitScalaFuture =
81 Patterns.ask(masterTxActor, new SubmitRequest(), askTimeout);
83 LOG.trace("{}: Commit {} via NETCONF", id);
85 final SettableFuture<CommitInfo> settableFuture = SettableFuture.create();
86 submitScalaFuture.onComplete(new OnComplete<Object>() {
88 public void onComplete(final Throwable failure, final Object success) throws Throwable {
89 if (failure != null) { // ask timeout
90 settableFuture.setException(newTransactionCommitFailedException(
91 NetconfTopologyUtils.createMasterIsDownException(id), identifier));
94 if (success instanceof Throwable) {
95 settableFuture.setException(newTransactionCommitFailedException((Throwable) success, identifier));
97 if (success instanceof SubmitFailedReply) {
98 LOG.error("{}: Transaction was not submitted because already closed.", id);
99 settableFuture.setException(newTransactionCommitFailedException(
100 ((SubmitFailedReply) success).getThrowable(), identifier));
104 settableFuture.set(CommitInfo.empty());
107 }, actorSystem.dispatcher());
109 return FluentFuture.from(settableFuture);
112 private static TransactionCommitFailedException newTransactionCommitFailedException(final Throwable failure,
113 final Object identifier) {
114 return new TransactionCommitFailedException(
115 String.format("Commit of transaction %s failed", identifier), failure);
118 public void delete(final LogicalDatastoreType store, final YangInstanceIdentifier identifier) {
119 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
120 LOG.trace("{}: Delete {} via NETCONF: {}", id, store, identifier);
121 masterTxActor.tell(new DeleteRequest(store, identifier), ActorRef.noSender());
124 public void put(final LogicalDatastoreType store, final YangInstanceIdentifier path,
125 final NormalizedNode<?, ?> data, final Object identifier) {
126 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
127 final NormalizedNodeMessage msg = new NormalizedNodeMessage(path, data);
128 LOG.trace("{}: Put {} via NETCONF: {} with payload {}", id, store, path, data);
129 masterTxActor.tell(new PutRequest(store, msg), ActorRef.noSender());
132 public void merge(final LogicalDatastoreType store, final YangInstanceIdentifier path,
133 final NormalizedNode<?, ?> data, final Object identifier) {
134 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
135 final NormalizedNodeMessage msg = new NormalizedNodeMessage(path, data);
136 LOG.trace("{}: Merge {} via NETCONF: {} with payload {}", id, store, path, data);
137 masterTxActor.tell(new MergeRequest(store, msg), ActorRef.noSender());