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.Function;
17 import com.google.common.base.Preconditions;
18 import com.google.common.util.concurrent.CheckedFuture;
19 import com.google.common.util.concurrent.Futures;
20 import com.google.common.util.concurrent.ListenableFuture;
21 import com.google.common.util.concurrent.SettableFuture;
22 import java.util.concurrent.atomic.AtomicBoolean;
23 import javax.annotation.Nullable;
24 import org.opendaylight.controller.md.sal.common.api.TransactionStatus;
25 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
26 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
27 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
28 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
29 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
30 import org.opendaylight.netconf.topology.singleton.messages.NormalizedNodeMessage;
31 import org.opendaylight.netconf.topology.singleton.messages.transactions.CancelRequest;
32 import org.opendaylight.netconf.topology.singleton.messages.transactions.DeleteRequest;
33 import org.opendaylight.netconf.topology.singleton.messages.transactions.MergeRequest;
34 import org.opendaylight.netconf.topology.singleton.messages.transactions.PutRequest;
35 import org.opendaylight.netconf.topology.singleton.messages.transactions.SubmitFailedReply;
36 import org.opendaylight.netconf.topology.singleton.messages.transactions.SubmitRequest;
37 import org.opendaylight.yangtools.yang.common.RpcResult;
38 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
39 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
40 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
41 import org.slf4j.Logger;
42 import org.slf4j.LoggerFactory;
43 import scala.concurrent.Await;
44 import scala.concurrent.Future;
47 * ProxyWriteTransaction uses provided {@link ActorRef} to delegate method calls to master
48 * {@link org.opendaylight.netconf.topology.singleton.impl.actors.WriteTransactionActor}.
50 public class ProxyWriteTransaction implements DOMDataWriteTransaction {
52 private static final Logger LOG = LoggerFactory.getLogger(ProxyWriteTransaction.class);
54 private final ActorRef masterTxActor;
55 private final RemoteDeviceId id;
56 private final ActorSystem actorSystem;
57 private final AtomicBoolean opened = new AtomicBoolean(true);
58 private final Timeout askTimeout;
61 * @param masterTxActor {@link org.opendaylight.netconf.topology.singleton.impl.actors.WriteTransactionActor} ref
63 * @param actorSystem system
66 public ProxyWriteTransaction(final ActorRef masterTxActor, final RemoteDeviceId id, final ActorSystem actorSystem,
67 final Timeout askTimeout) {
68 this.masterTxActor = masterTxActor;
70 this.actorSystem = actorSystem;
71 this.askTimeout = askTimeout;
75 public boolean cancel() {
76 if (!opened.compareAndSet(true, false)) {
79 final Future<Object> cancelScalaFuture =
80 Patterns.ask(masterTxActor, new CancelRequest(), askTimeout);
82 LOG.trace("{}: Cancel {} via NETCONF", id);
85 // here must be Await because AsyncWriteTransaction do not return future
86 return (boolean) Await.result(cancelScalaFuture, askTimeout.duration());
87 } catch (final Exception e) {
93 public CheckedFuture<Void, TransactionCommitFailedException> submit() {
94 if (!opened.compareAndSet(true, false)) {
95 throw new IllegalStateException(id + ": Transaction" + getIdentifier() + " is closed");
97 final Future<Object> submitScalaFuture =
98 Patterns.ask(masterTxActor, new SubmitRequest(), askTimeout);
100 LOG.trace("{}: Submit {} via NETCONF", id);
102 final SettableFuture<Void> settableFuture = SettableFuture.create();
103 submitScalaFuture.onComplete(new OnComplete<Object>() {
105 public void onComplete(final Throwable failure, final Object success) throws Throwable {
106 if (failure != null) { // ask timeout
107 final Exception exception = NetconfTopologyUtils.createMasterIsDownException(id);
108 settableFuture.setException(exception);
111 if (success instanceof Throwable) {
112 settableFuture.setException((Throwable) success);
114 if (success instanceof SubmitFailedReply) {
115 LOG.error("{}: Transaction was not submitted because already closed.", id);
117 settableFuture.set(null);
120 }, actorSystem.dispatcher());
122 return Futures.makeChecked(settableFuture, new Function<Exception, TransactionCommitFailedException>() {
125 public TransactionCommitFailedException apply(@Nullable final Exception input) {
126 final String message = "Submit of transaction " + getIdentifier() + " failed";
127 return new TransactionCommitFailedException(message, input);
133 public ListenableFuture<RpcResult<TransactionStatus>> commit() {
134 LOG.trace("{}: Commit", id);
136 final CheckedFuture<Void, TransactionCommitFailedException> submit = submit();
137 return Futures.transform(submit, new Function<Void, RpcResult<TransactionStatus>>() {
140 public RpcResult<TransactionStatus> apply(@Nullable final Void input) {
141 return RpcResultBuilder.success(TransactionStatus.SUBMITED).build();
147 public void delete(final LogicalDatastoreType store, final YangInstanceIdentifier identifier) {
148 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, getIdentifier());
149 LOG.trace("{}: Delete {} via NETCONF: {}", id, store, identifier);
150 masterTxActor.tell(new DeleteRequest(store, identifier), ActorRef.noSender());
154 public void put(final LogicalDatastoreType store, final YangInstanceIdentifier identifier,
155 final NormalizedNode<?, ?> data) {
156 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, getIdentifier());
157 final NormalizedNodeMessage msg = new NormalizedNodeMessage(identifier, data);
158 LOG.trace("{}: Put {} via NETCONF: {} with payload {}", id, store, identifier, data);
159 masterTxActor.tell(new PutRequest(store, msg), ActorRef.noSender());
163 public void merge(final LogicalDatastoreType store, final YangInstanceIdentifier identifier,
164 final NormalizedNode<?, ?> data) {
165 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, getIdentifier());
166 final NormalizedNodeMessage msg = new NormalizedNodeMessage(identifier, data);
167 LOG.trace("{}: Merge {} via NETCONF: {} with payload {}", id, store, identifier, data);
168 masterTxActor.tell(new MergeRequest(store, msg), ActorRef.noSender());
172 public Object getIdentifier() {