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 Exception exception = new DocumentedException(id + ":Master is down. Please try again.",
94 DocumentedException.ErrorType.APPLICATION, DocumentedException.ErrorTag.OPERATION_FAILED,
95 DocumentedException.ErrorSeverity.WARNING);
96 promise.failure(exception);
99 if (success instanceof Throwable) { // Error sended by master
100 promise.failure((Throwable) success);
103 if (success instanceof EmptyReadResponse) {
104 promise.success(Optional.absent());
107 promise.success(Optional.of((NormalizedNodeMessage) success));
109 }, actorSystem.dispatcher());
111 return promise.future();
115 public Future<Boolean> exists(final LogicalDatastoreType store, final YangInstanceIdentifier path) {
116 final Future<Object> existsScalaFuture =
117 Patterns.ask(masterContextRef, new ExistsRequest(store, path), actorResponseWaitTime);
119 LOG.trace("{}: Exists {} via NETCONF: {}", id, store, path);
121 final DefaultPromise<Boolean> promise = new DefaultPromise<>();
122 existsScalaFuture.onComplete(new OnComplete<Object>() {
124 public void onComplete(final Throwable failure, final Object success) throws Throwable {
125 if (failure != null) { // ask timeout
126 Exception exception = new DocumentedException(id + ":Master is down. Please try again.",
127 DocumentedException.ErrorType.APPLICATION, DocumentedException.ErrorTag.OPERATION_FAILED,
128 DocumentedException.ErrorSeverity.WARNING);
129 promise.failure(exception);
132 if (success instanceof Throwable) {
133 promise.failure((Throwable) success);
136 promise.success((Boolean) success);
138 }, actorSystem.dispatcher());
139 return promise.future();
143 public void put(final LogicalDatastoreType store, final NormalizedNodeMessage data) {
144 LOG.trace("{}: Write {} via NETCONF: {} with payload {}", id, store, data.getIdentifier(), data.getNode());
146 masterContextRef.tell(new PutRequest(store, data), ActorRef.noSender());
151 public void merge(final LogicalDatastoreType store, final NormalizedNodeMessage data) {
152 LOG.trace("{}: Merge {} via NETCONF: {} with payload {}", id, store, data.getIdentifier(), data.getNode());
154 masterContextRef.tell(new MergeRequest(store, data), ActorRef.noSender());
158 public void delete(final LogicalDatastoreType store, final YangInstanceIdentifier path) {
159 LOG.trace("{}: Delete {} via NETCONF: {}", id, store, path);
161 masterContextRef.tell(new DeleteRequest(store, path), ActorRef.noSender());
165 public boolean cancel() {
166 final Future<Object> cancelScalaFuture =
167 Patterns.ask(masterContextRef, new CancelRequest(), actorResponseWaitTime);
169 LOG.trace("{}: Cancel {} via NETCONF", id);
172 // here must be Await because AsyncWriteTransaction do not return future
173 return (boolean) Await.result(cancelScalaFuture, actorResponseWaitTime.duration());
174 } catch (Exception e) {
180 public Future<Void> submit() {
181 final Future<Object> submitScalaFuture =
182 Patterns.ask(masterContextRef, new SubmitRequest(), actorResponseWaitTime);
184 LOG.trace("{}: Submit {} via NETCONF", id);
186 final DefaultPromise<Void> promise = new DefaultPromise<>();
188 submitScalaFuture.onComplete(new OnComplete<Object>() {
190 public void onComplete(final Throwable failure, final Object success) throws Throwable {
191 if (failure != null) { // ask timeout
192 Exception exception = new DocumentedException(id + ":Master is down. Please try again.",
193 DocumentedException.ErrorType.APPLICATION, DocumentedException.ErrorTag.OPERATION_FAILED,
194 DocumentedException.ErrorSeverity.WARNING);
195 promise.failure(exception);
198 if (success instanceof Throwable) {
199 promise.failure((Throwable) success);
201 if (success instanceof SubmitFailedReply) {
202 LOG.error("{}: Transaction was not submitted because already closed.", id);
204 promise.success(null);
207 }, actorSystem.dispatcher());
209 return promise.future();