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.SettableFuture;
21 import java.util.concurrent.atomic.AtomicBoolean;
22 import javax.annotation.Nullable;
23 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
24 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
25 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
26 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologyUtils;
27 import org.opendaylight.netconf.topology.singleton.messages.NormalizedNodeMessage;
28 import org.opendaylight.netconf.topology.singleton.messages.transactions.CancelRequest;
29 import org.opendaylight.netconf.topology.singleton.messages.transactions.DeleteRequest;
30 import org.opendaylight.netconf.topology.singleton.messages.transactions.MergeRequest;
31 import org.opendaylight.netconf.topology.singleton.messages.transactions.PutRequest;
32 import org.opendaylight.netconf.topology.singleton.messages.transactions.SubmitFailedReply;
33 import org.opendaylight.netconf.topology.singleton.messages.transactions.SubmitRequest;
34 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
35 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
36 import org.slf4j.Logger;
37 import org.slf4j.LoggerFactory;
38 import scala.concurrent.Await;
39 import scala.concurrent.Future;
41 public class ProxyWriteAdapter {
43 private static final Logger LOG = LoggerFactory.getLogger(ProxyWriteAdapter.class);
45 private final ActorRef masterTxActor;
46 private final RemoteDeviceId id;
47 private final ActorSystem actorSystem;
48 private final AtomicBoolean opened = new AtomicBoolean(true);
49 private final Timeout askTimeout;
51 public ProxyWriteAdapter(final ActorRef masterTxActor, final RemoteDeviceId id, final ActorSystem actorSystem,
52 final Timeout askTimeout) {
53 this.masterTxActor = masterTxActor;
55 this.actorSystem = actorSystem;
56 this.askTimeout = askTimeout;
59 @SuppressWarnings("checkstyle:IllegalCatch")
60 public boolean cancel() {
61 if (!opened.compareAndSet(true, false)) {
64 final Future<Object> cancelScalaFuture =
65 Patterns.ask(masterTxActor, new CancelRequest(), askTimeout);
67 LOG.trace("{}: Cancel {} via NETCONF", id);
70 // here must be Await because AsyncWriteTransaction do not return future
71 return (boolean) Await.result(cancelScalaFuture, askTimeout.duration());
72 } catch (final Exception e) {
77 public CheckedFuture<Void, TransactionCommitFailedException> submit(final Object identifier) {
78 if (!opened.compareAndSet(true, false)) {
79 throw new IllegalStateException(id + ": Transaction" + identifier + " is closed");
81 final Future<Object> submitScalaFuture =
82 Patterns.ask(masterTxActor, new SubmitRequest(), askTimeout);
84 LOG.trace("{}: Submit {} via NETCONF", id);
86 final SettableFuture<Void> settableFuture = SettableFuture.create();
87 submitScalaFuture.onComplete(new OnComplete<Object>() {
89 public void onComplete(final Throwable failure, final Object success) throws Throwable {
90 if (failure != null) { // ask timeout
91 final Exception exception = NetconfTopologyUtils.createMasterIsDownException(id);
92 settableFuture.setException(exception);
95 if (success instanceof Throwable) {
96 settableFuture.setException((Throwable) success);
98 if (success instanceof SubmitFailedReply) {
99 LOG.error("{}: Transaction was not submitted because already closed.", id);
100 settableFuture.setException(((SubmitFailedReply) success).getThrowable());
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 void delete(final LogicalDatastoreType store, final YangInstanceIdentifier identifier) {
120 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
121 LOG.trace("{}: Delete {} via NETCONF: {}", id, store, identifier);
122 masterTxActor.tell(new DeleteRequest(store, identifier), ActorRef.noSender());
125 public void put(final LogicalDatastoreType store, final YangInstanceIdentifier path,
126 final NormalizedNode<?, ?> data, final Object identifier) {
127 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
128 final NormalizedNodeMessage msg = new NormalizedNodeMessage(path, data);
129 LOG.trace("{}: Put {} via NETCONF: {} with payload {}", id, store, path, data);
130 masterTxActor.tell(new PutRequest(store, msg), ActorRef.noSender());
133 public void merge(final LogicalDatastoreType store, final YangInstanceIdentifier path,
134 final NormalizedNode<?, ?> data, final Object identifier) {
135 Preconditions.checkState(opened.get(), "%s: Transaction was closed %s", id, identifier);
136 final NormalizedNodeMessage msg = new NormalizedNodeMessage(path, data);
137 LOG.trace("{}: Merge {} via NETCONF: {} with payload {}", id, store, path, data);
138 masterTxActor.tell(new MergeRequest(store, msg), ActorRef.noSender());