2 * Copyright (c) 2016 Cisco Systems, Inc. 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.Optional;
17 import com.google.common.base.Throwables;
18 import org.opendaylight.controller.config.util.xml.DocumentedException;
19 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
20 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
21 import org.opendaylight.netconf.topology.singleton.api.NetconfDOMTransaction;
22 import org.opendaylight.netconf.topology.singleton.messages.NormalizedNodeMessage;
23 import org.opendaylight.netconf.topology.singleton.messages.transactions.CancelRequest;
24 import org.opendaylight.netconf.topology.singleton.messages.transactions.DeleteRequest;
25 import org.opendaylight.netconf.topology.singleton.messages.transactions.EmptyReadResponse;
26 import org.opendaylight.netconf.topology.singleton.messages.transactions.ExistsRequest;
27 import org.opendaylight.netconf.topology.singleton.messages.transactions.MergeRequest;
28 import org.opendaylight.netconf.topology.singleton.messages.transactions.OpenTransaction;
29 import org.opendaylight.netconf.topology.singleton.messages.transactions.PutRequest;
30 import org.opendaylight.netconf.topology.singleton.messages.transactions.ReadRequest;
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.slf4j.Logger;
35 import org.slf4j.LoggerFactory;
36 import scala.concurrent.Await;
37 import scala.concurrent.Future;
38 import scala.concurrent.impl.Promise.DefaultPromise;
41 public class NetconfProxyDOMTransaction implements NetconfDOMTransaction {
43 private static final Logger LOG = LoggerFactory.getLogger(NetconfProxyDOMTransaction.class);
45 private final RemoteDeviceId id;
46 private final ActorSystem actorSystem;
47 private final ActorRef masterContextRef;
48 private final Timeout actorResponseWaitTime;
50 public NetconfProxyDOMTransaction(final RemoteDeviceId id,
51 final ActorSystem actorSystem,
52 final ActorRef masterContextRef,
53 final Timeout actorResponseWaitTime) {
55 this.actorSystem = actorSystem;
56 this.masterContextRef = masterContextRef;
57 this.actorResponseWaitTime = actorResponseWaitTime;
61 public void openTransaction() {
62 // TODO we can do some checking for already opened transaction also
63 // here in this class. We can track open transaction at least for this
65 LOG.debug("{}: Requesting leader {} to open new transaction", id, masterContextRef);
66 final Future<Object> openTxFuture =
67 Patterns.ask(masterContextRef, new OpenTransaction(), actorResponseWaitTime);
69 // we have to wait here so we can see if tx can be opened
70 Await.result(openTxFuture, actorResponseWaitTime.duration());
71 LOG.debug("{}: New transaction opened successfully", id);
72 } catch (final Exception e) {
73 LOG.error("{}: Failed to open new transaction", id, e);
74 Throwables.propagate(e);
79 public Future<Optional<NormalizedNodeMessage>> read(final LogicalDatastoreType store,
80 final YangInstanceIdentifier path) {
82 final Future<Object> readScalaFuture =
83 Patterns.ask(masterContextRef, new ReadRequest(store, path), actorResponseWaitTime);
85 LOG.trace("{}: Read {} via NETCONF: {}", id, store, path);
87 final DefaultPromise<Optional<NormalizedNodeMessage>> promise = new DefaultPromise<>();
89 readScalaFuture.onComplete(new OnComplete<Object>() {
91 public void onComplete(final Throwable failure, final Object success) throws Throwable {
92 if (failure != null) { // ask timeout
93 final Exception exception = new DocumentedException(
94 id + ":Master is down. Please try again.",
95 DocumentedException.ErrorType.TRANSPORT,
96 DocumentedException.ErrorTag.RESOURCE_DENIED,
97 DocumentedException.ErrorSeverity.ERROR);
98 promise.failure(exception);
101 if (success instanceof Throwable) { // Error sended by master
102 promise.failure((Throwable) success);
105 if (success instanceof EmptyReadResponse) {
106 promise.success(Optional.absent());
109 promise.success(Optional.of((NormalizedNodeMessage) success));
111 }, actorSystem.dispatcher());
113 return promise.future();
117 public Future<Boolean> exists(final LogicalDatastoreType store, final YangInstanceIdentifier path) {
118 final Future<Object> existsScalaFuture =
119 Patterns.ask(masterContextRef, new ExistsRequest(store, path), actorResponseWaitTime);
121 LOG.trace("{}: Exists {} via NETCONF: {}", id, store, path);
123 final DefaultPromise<Boolean> promise = new DefaultPromise<>();
124 existsScalaFuture.onComplete(new OnComplete<Object>() {
126 public void onComplete(final Throwable failure, final Object success) throws Throwable {
127 if (failure != null) { // ask timeout
128 final Exception exception = new DocumentedException(
129 id + ":Master is down. Please try again.",
130 DocumentedException.ErrorType.TRANSPORT,
131 DocumentedException.ErrorTag.RESOURCE_DENIED,
132 DocumentedException.ErrorSeverity.ERROR);
133 promise.failure(exception);
136 if (success instanceof Throwable) {
137 promise.failure((Throwable) success);
140 promise.success((Boolean) success);
142 }, actorSystem.dispatcher());
143 return promise.future();
147 public void put(final LogicalDatastoreType store, final NormalizedNodeMessage data) {
148 LOG.trace("{}: Write {} via NETCONF: {} with payload {}", id, store, data.getIdentifier(), data.getNode());
150 masterContextRef.tell(new PutRequest(store, data), ActorRef.noSender());
154 public void merge(final LogicalDatastoreType store, final NormalizedNodeMessage data) {
155 LOG.trace("{}: Merge {} via NETCONF: {} with payload {}", id, store, data.getIdentifier(), data.getNode());
157 masterContextRef.tell(new MergeRequest(store, data), ActorRef.noSender());
161 public void delete(final LogicalDatastoreType store, final YangInstanceIdentifier path) {
162 LOG.trace("{}: Delete {} via NETCONF: {}", id, store, path);
164 masterContextRef.tell(new DeleteRequest(store, path), ActorRef.noSender());
168 public boolean cancel() {
169 final Future<Object> cancelScalaFuture =
170 Patterns.ask(masterContextRef, new CancelRequest(), actorResponseWaitTime);
172 LOG.trace("{}: Cancel {} via NETCONF", id);
175 // here must be Await because AsyncWriteTransaction do not return future
176 return (boolean) Await.result(cancelScalaFuture, actorResponseWaitTime.duration());
177 } catch (Exception e) {
183 public Future<Void> submit() {
184 final Future<Object> submitScalaFuture =
185 Patterns.ask(masterContextRef, new SubmitRequest(), actorResponseWaitTime);
187 LOG.trace("{}: Submit {} via NETCONF", id);
189 final DefaultPromise<Void> promise = new DefaultPromise<>();
191 submitScalaFuture.onComplete(new OnComplete<Object>() {
193 public void onComplete(final Throwable failure, final Object success) throws Throwable {
194 if (failure != null) { // ask timeout
195 final Exception exception = new DocumentedException(
196 id + ":Master is down. Please try again.",
197 DocumentedException.ErrorType.TRANSPORT,
198 DocumentedException.ErrorTag.RESOURCE_DENIED,
199 DocumentedException.ErrorSeverity.ERROR);
200 promise.failure(exception);
203 if (success instanceof Throwable) {
204 promise.failure((Throwable) success);
206 if (success instanceof SubmitFailedReply) {
207 LOG.error("{}: Transaction was not submitted because already closed.", id);
209 promise.success(null);
212 }, actorSystem.dispatcher());
214 return promise.future();