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.netconf.sal.connect.util.RemoteDeviceId;
28 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
29 import org.opendaylight.netconf.topology.singleton.messages.NormalizedNodeMessage;
30 import org.opendaylight.netconf.topology.singleton.messages.transactions.CancelRequest;
31 import org.opendaylight.netconf.topology.singleton.messages.transactions.DeleteRequest;
32 import org.opendaylight.netconf.topology.singleton.messages.transactions.MergeRequest;
33 import org.opendaylight.netconf.topology.singleton.messages.transactions.PutRequest;
34 import org.opendaylight.netconf.topology.singleton.messages.transactions.SubmitFailedReply;
35 import org.opendaylight.netconf.topology.singleton.messages.transactions.SubmitRequest;
36 import org.opendaylight.yangtools.yang.common.RpcResult;
37 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
38 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
39 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
40 import org.slf4j.Logger;
41 import org.slf4j.LoggerFactory;
42 import scala.concurrent.Await;
43 import scala.concurrent.Future;
45 public class ProxyWriteAdapter {
47 private static final Logger LOG = LoggerFactory.getLogger(ProxyWriteAdapter.class);
49 private final ActorRef masterTxActor;
50 private final RemoteDeviceId id;
51 private final ActorSystem actorSystem;
52 private final AtomicBoolean opened = new AtomicBoolean(true);
53 private final Timeout askTimeout;
55 public ProxyWriteAdapter(final ActorRef masterTxActor, final RemoteDeviceId id, final ActorSystem actorSystem,
56 final Timeout askTimeout) {
57 this.masterTxActor = masterTxActor;
59 this.actorSystem = actorSystem;
60 this.askTimeout = askTimeout;
63 public boolean cancel() {
64 if (!opened.compareAndSet(true, false)) {
67 final Future<Object> cancelScalaFuture =
68 Patterns.ask(masterTxActor, new CancelRequest(), askTimeout);
70 LOG.trace("{}: Cancel {} via NETCONF", id);
73 // here must be Await because AsyncWriteTransaction do not return future
74 return (boolean) Await.result(cancelScalaFuture, askTimeout.duration());
75 } catch (final Exception e) {
80 public CheckedFuture<Void, TransactionCommitFailedException> submit(final Object identifier) {
81 if (!opened.compareAndSet(true, false)) {
82 throw new IllegalStateException(id + ": Transaction" + identifier + " is closed");
84 final Future<Object> submitScalaFuture =
85 Patterns.ask(masterTxActor, new SubmitRequest(), askTimeout);
87 LOG.trace("{}: Submit {} via NETCONF", id);
89 final SettableFuture<Void> settableFuture = SettableFuture.create();
90 submitScalaFuture.onComplete(new OnComplete<Object>() {
92 public void onComplete(final Throwable failure, final Object success) throws Throwable {
93 if (failure != null) { // ask timeout
94 final Exception exception = NetconfTopologyUtils.createMasterIsDownException(id);
95 settableFuture.setException(exception);
98 if (success instanceof Throwable) {
99 settableFuture.setException((Throwable) success);
101 if (success instanceof SubmitFailedReply) {
102 LOG.error("{}: Transaction was not submitted because already closed.", id);
104 settableFuture.set(null);
107 }, actorSystem.dispatcher());
109 return Futures.makeChecked(settableFuture, new Function<Exception, TransactionCommitFailedException>() {
112 public TransactionCommitFailedException apply(@Nullable final Exception input) {
113 final String message = "Submit of transaction " + identifier + " failed";
114 return new TransactionCommitFailedException(message, input);
119 public ListenableFuture<RpcResult<TransactionStatus>> commit(final Object identifier) {
120 LOG.trace("{}: Commit", id);
122 final CheckedFuture<Void, TransactionCommitFailedException> submit = submit(identifier);
123 return Futures.transform(submit, new Function<Void, RpcResult<TransactionStatus>>() {
126 public RpcResult<TransactionStatus> apply(@Nullable final Void input) {
127 return RpcResultBuilder.success(TransactionStatus.SUBMITED).build();
132 public void delete(final LogicalDatastoreType store, final YangInstanceIdentifier identifier) {
133 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
134 LOG.trace("{}: Delete {} via NETCONF: {}", id, store, identifier);
135 masterTxActor.tell(new DeleteRequest(store, identifier), ActorRef.noSender());
138 public void put(final LogicalDatastoreType store, final YangInstanceIdentifier path,
139 final NormalizedNode<?, ?> data, final Object identifier) {
140 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
141 final NormalizedNodeMessage msg = new NormalizedNodeMessage(path, data);
142 LOG.trace("{}: Put {} via NETCONF: {} with payload {}", id, store, path, data);
143 masterTxActor.tell(new PutRequest(store, msg), ActorRef.noSender());
146 public void merge(final LogicalDatastoreType store, final YangInstanceIdentifier path,
147 final NormalizedNode<?, ?> data, final Object identifier) {
148 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
149 final NormalizedNodeMessage msg = new NormalizedNodeMessage(path, data);
150 LOG.trace("{}: Merge {} via NETCONF: {} with payload {}", id, store, path, data);
151 masterTxActor.tell(new MergeRequest(store, msg), ActorRef.noSender());