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 @SuppressWarnings("checkstyle:IllegalCatch")
64 public boolean cancel() {
65 if (!opened.compareAndSet(true, false)) {
68 final Future<Object> cancelScalaFuture =
69 Patterns.ask(masterTxActor, new CancelRequest(), askTimeout);
71 LOG.trace("{}: Cancel {} via NETCONF", id);
74 // here must be Await because AsyncWriteTransaction do not return future
75 return (boolean) Await.result(cancelScalaFuture, askTimeout.duration());
76 } catch (final Exception e) {
81 public CheckedFuture<Void, TransactionCommitFailedException> submit(final Object identifier) {
82 if (!opened.compareAndSet(true, false)) {
83 throw new IllegalStateException(id + ": Transaction" + identifier + " is closed");
85 final Future<Object> submitScalaFuture =
86 Patterns.ask(masterTxActor, new SubmitRequest(), askTimeout);
88 LOG.trace("{}: Submit {} via NETCONF", id);
90 final SettableFuture<Void> settableFuture = SettableFuture.create();
91 submitScalaFuture.onComplete(new OnComplete<Object>() {
93 public void onComplete(final Throwable failure, final Object success) throws Throwable {
94 if (failure != null) { // ask timeout
95 final Exception exception = NetconfTopologyUtils.createMasterIsDownException(id);
96 settableFuture.setException(exception);
99 if (success instanceof Throwable) {
100 settableFuture.setException((Throwable) success);
102 if (success instanceof SubmitFailedReply) {
103 LOG.error("{}: Transaction was not submitted because already closed.", id);
105 settableFuture.set(null);
108 }, actorSystem.dispatcher());
110 return Futures.makeChecked(settableFuture, new Function<Exception, TransactionCommitFailedException>() {
113 public TransactionCommitFailedException apply(@Nullable final Exception input) {
114 final String message = "Submit of transaction " + identifier + " failed";
115 return new TransactionCommitFailedException(message, input);
120 public ListenableFuture<RpcResult<TransactionStatus>> commit(final Object identifier) {
121 LOG.trace("{}: Commit", id);
123 final CheckedFuture<Void, TransactionCommitFailedException> submit = submit(identifier);
124 return Futures.transform(submit, new Function<Void, RpcResult<TransactionStatus>>() {
127 public RpcResult<TransactionStatus> apply(@Nullable final Void input) {
128 return RpcResultBuilder.success(TransactionStatus.SUBMITED).build();
133 public void delete(final LogicalDatastoreType store, final YangInstanceIdentifier identifier) {
134 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
135 LOG.trace("{}: Delete {} via NETCONF: {}", id, store, identifier);
136 masterTxActor.tell(new DeleteRequest(store, identifier), ActorRef.noSender());
139 public void put(final LogicalDatastoreType store, final YangInstanceIdentifier path,
140 final NormalizedNode<?, ?> data, final Object identifier) {
141 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
142 final NormalizedNodeMessage msg = new NormalizedNodeMessage(path, data);
143 LOG.trace("{}: Put {} via NETCONF: {} with payload {}", id, store, path, data);
144 masterTxActor.tell(new PutRequest(store, msg), ActorRef.noSender());
147 public void merge(final LogicalDatastoreType store, final YangInstanceIdentifier path,
148 final NormalizedNode<?, ?> data, final Object identifier) {
149 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
150 final NormalizedNodeMessage msg = new NormalizedNodeMessage(path, data);
151 LOG.trace("{}: Merge {} via NETCONF: {} with payload {}", id, store, path, data);
152 masterTxActor.tell(new MergeRequest(store, msg), ActorRef.noSender());