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.actors;
11 import akka.actor.ActorRef;
12 import akka.actor.Props;
13 import akka.actor.UntypedActor;
14 import akka.util.Timeout;
15 import com.google.common.util.concurrent.CheckedFuture;
16 import com.google.common.util.concurrent.FutureCallback;
17 import com.google.common.util.concurrent.Futures;
18 import java.io.IOException;
19 import java.util.List;
20 import javax.annotation.Nonnull;
21 import javax.annotation.Nullable;
22 import org.opendaylight.controller.cluster.schema.provider.RemoteYangTextSourceProvider;
23 import org.opendaylight.controller.cluster.schema.provider.impl.RemoteSchemaProvider;
24 import org.opendaylight.controller.cluster.schema.provider.impl.YangTextSchemaSourceSerializationProxy;
25 import org.opendaylight.controller.md.sal.dom.api.DOMDataBroker;
26 import org.opendaylight.controller.md.sal.dom.api.DOMDataReadOnlyTransaction;
27 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
28 import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
29 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
30 import org.opendaylight.controller.md.sal.dom.api.DOMRpcService;
31 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
32 import org.opendaylight.netconf.topology.singleton.impl.ProxyDOMRpcService;
33 import org.opendaylight.netconf.topology.singleton.impl.ProxyYangTextSourceProvider;
34 import org.opendaylight.netconf.topology.singleton.impl.SlaveSalFacade;
35 import org.opendaylight.netconf.topology.singleton.impl.utils.NetconfTopologySetup;
36 import org.opendaylight.netconf.topology.singleton.messages.AskForMasterMountPoint;
37 import org.opendaylight.netconf.topology.singleton.messages.CreateInitialMasterActorData;
38 import org.opendaylight.netconf.topology.singleton.messages.MasterActorDataInitialized;
39 import org.opendaylight.netconf.topology.singleton.messages.NormalizedNodeMessage;
40 import org.opendaylight.netconf.topology.singleton.messages.RefreshSetupMasterActorData;
41 import org.opendaylight.netconf.topology.singleton.messages.RegisterMountPoint;
42 import org.opendaylight.netconf.topology.singleton.messages.UnregisterSlaveMountPoint;
43 import org.opendaylight.netconf.topology.singleton.messages.YangTextSchemaSourceRequest;
44 import org.opendaylight.netconf.topology.singleton.messages.rpc.InvokeRpcMessage;
45 import org.opendaylight.netconf.topology.singleton.messages.rpc.InvokeRpcMessageReply;
46 import org.opendaylight.netconf.topology.singleton.messages.transactions.EmptyResultResponse;
47 import org.opendaylight.netconf.topology.singleton.messages.transactions.NewReadTransactionReply;
48 import org.opendaylight.netconf.topology.singleton.messages.transactions.NewReadTransactionRequest;
49 import org.opendaylight.netconf.topology.singleton.messages.transactions.NewWriteTransactionReply;
50 import org.opendaylight.netconf.topology.singleton.messages.transactions.NewWriteTransactionRequest;
51 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
52 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
53 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
54 import org.opendaylight.yangtools.yang.model.repo.api.SchemaContextFactory;
55 import org.opendaylight.yangtools.yang.model.repo.api.SchemaRepository;
56 import org.opendaylight.yangtools.yang.model.repo.api.SchemaResolutionException;
57 import org.opendaylight.yangtools.yang.model.repo.api.SchemaSourceException;
58 import org.opendaylight.yangtools.yang.model.repo.api.SchemaSourceFilter;
59 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
60 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
61 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
62 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistry;
63 import org.slf4j.Logger;
64 import org.slf4j.LoggerFactory;
66 public class NetconfNodeActor extends UntypedActor {
68 private static final Logger LOG = LoggerFactory.getLogger(NetconfNodeActor.class);
70 private NetconfTopologySetup setup;
71 private RemoteDeviceId id;
72 private final SchemaSourceRegistry schemaRegistry;
73 private final SchemaRepository schemaRepository;
75 private List<SourceIdentifier> sourceIdentifiers;
76 private DOMRpcService deviceRpc;
77 private SlaveSalFacade slaveSalManager;
78 private final Timeout actorResponseWaitTime;
79 private DOMDataBroker deviceDataBroker;
80 //readTxActor can be shared
81 private ActorRef readTxActor;
83 public static Props props(final NetconfTopologySetup setup,
84 final RemoteDeviceId id, final SchemaSourceRegistry schemaRegistry,
85 final SchemaRepository schemaRepository, final Timeout actorResponseWaitTime) {
86 return Props.create(NetconfNodeActor.class, () ->
87 new NetconfNodeActor(setup, id, schemaRegistry, schemaRepository, actorResponseWaitTime));
90 private NetconfNodeActor(final NetconfTopologySetup setup,
91 final RemoteDeviceId id, final SchemaSourceRegistry schemaRegistry,
92 final SchemaRepository schemaRepository, final Timeout actorResponseWaitTime) {
95 this.schemaRegistry = schemaRegistry;
96 this.schemaRepository = schemaRepository;
97 this.actorResponseWaitTime = actorResponseWaitTime;
101 public void onReceive(final Object message) throws Exception {
102 if (message instanceof CreateInitialMasterActorData) { // master
104 final CreateInitialMasterActorData masterActorData = (CreateInitialMasterActorData) message;
105 sourceIdentifiers = masterActorData.getSourceIndentifiers();
106 this.deviceDataBroker = masterActorData.getDeviceDataBroker();
107 final DOMDataReadOnlyTransaction tx = deviceDataBroker.newReadOnlyTransaction();
108 readTxActor = context().actorOf(ReadTransactionActor.props(tx));
109 this.deviceRpc = masterActorData.getDeviceRpc();
111 sender().tell(new MasterActorDataInitialized(), self());
113 LOG.debug("{}: Master is ready.", id);
115 } else if (message instanceof RefreshSetupMasterActorData) {
116 setup = ((RefreshSetupMasterActorData) message).getNetconfTopologyDeviceSetup();
117 id = ((RefreshSetupMasterActorData) message).getRemoteDeviceId();
118 sender().tell(new MasterActorDataInitialized(), self());
119 } else if (message instanceof AskForMasterMountPoint) { // master
120 // only master contains reference to deviceDataBroker
121 if (deviceDataBroker != null) {
122 getSender().tell(new RegisterMountPoint(sourceIdentifiers), getSelf());
125 } else if (message instanceof YangTextSchemaSourceRequest) { // master
127 final YangTextSchemaSourceRequest yangTextSchemaSourceRequest = (YangTextSchemaSourceRequest) message;
128 sendYangTextSchemaSourceProxy(yangTextSchemaSourceRequest.getSourceIdentifier(), sender());
130 } else if (message instanceof NewReadTransactionRequest) { // master
132 sender().tell(new NewReadTransactionReply(readTxActor), self());
134 } else if (message instanceof NewWriteTransactionRequest) { // master
136 final DOMDataWriteTransaction tx = deviceDataBroker.newWriteOnlyTransaction();
137 final ActorRef txActor = context().actorOf(WriteTransactionActor.props(tx));
138 sender().tell(new NewWriteTransactionReply(txActor), self());
139 } catch (final Throwable t) {
140 sender().tell(t, self());
143 } else if (message instanceof InvokeRpcMessage) { // master
145 final InvokeRpcMessage invokeRpcMessage = ((InvokeRpcMessage) message);
146 invokeSlaveRpc(invokeRpcMessage.getSchemaPath(), invokeRpcMessage.getNormalizedNodeMessage(), sender());
148 } else if (message instanceof RegisterMountPoint) { //slaves
150 sourceIdentifiers = ((RegisterMountPoint) message).getSourceIndentifiers();
151 registerSlaveMountPoint(getSender());
153 } else if (message instanceof UnregisterSlaveMountPoint) { //slaves
154 if (slaveSalManager != null) {
155 slaveSalManager.close();
156 slaveSalManager = null;
163 private void sendYangTextSchemaSourceProxy(final SourceIdentifier sourceIdentifier, final ActorRef sender) {
164 final CheckedFuture<YangTextSchemaSource, SchemaSourceException> yangTextSchemaSource =
165 schemaRepository.getSchemaSource(sourceIdentifier, YangTextSchemaSource.class);
167 Futures.addCallback(yangTextSchemaSource, new FutureCallback<YangTextSchemaSource>() {
169 public void onSuccess(final YangTextSchemaSource yangTextSchemaSource) {
171 sender.tell(new YangTextSchemaSourceSerializationProxy(yangTextSchemaSource), getSelf());
172 } catch (final IOException exception) {
173 sender.tell(exception.getCause(), getSelf());
178 public void onFailure(@Nonnull final Throwable throwable) {
179 sender.tell(throwable, getSelf());
184 private void invokeSlaveRpc(final SchemaPath schemaPath, final NormalizedNodeMessage normalizedNodeMessage,
185 final ActorRef recipient) {
187 final CheckedFuture<DOMRpcResult, DOMRpcException> rpcResult =
188 deviceRpc.invokeRpc(schemaPath, normalizedNodeMessage.getNode());
190 Futures.addCallback(rpcResult, new FutureCallback<DOMRpcResult>() {
192 public void onSuccess(@Nullable final DOMRpcResult domRpcResult) {
193 if (domRpcResult == null) {
194 recipient.tell(new EmptyResultResponse(), getSender());
197 NormalizedNodeMessage nodeMessageReply = null;
198 if (domRpcResult.getResult() != null) {
199 nodeMessageReply = new NormalizedNodeMessage(YangInstanceIdentifier.EMPTY,
200 domRpcResult.getResult());
202 recipient.tell(new InvokeRpcMessageReply(nodeMessageReply, domRpcResult.getErrors()), getSelf());
206 public void onFailure(@Nonnull final Throwable throwable) {
207 recipient.tell(throwable, getSelf());
212 private void registerSlaveMountPoint(final ActorRef masterReference) {
213 if (this.slaveSalManager != null) {
214 slaveSalManager.close();
216 slaveSalManager = new SlaveSalFacade(id, setup.getDomBroker(), setup.getActorSystem(), actorResponseWaitTime);
218 final CheckedFuture<SchemaContext, SchemaResolutionException> remoteSchemaContext =
219 getSchemaContext(masterReference);
220 final DOMRpcService deviceRpc = getDOMRpcService(masterReference);
222 Futures.addCallback(remoteSchemaContext, new FutureCallback<SchemaContext>() {
224 public void onSuccess(final SchemaContext result) {
225 LOG.info("{}: Schema context resolved: {}", id, result.getModules());
226 slaveSalManager.registerSlaveMountPoint(result, deviceRpc, masterReference);
230 public void onFailure(@Nonnull final Throwable throwable) {
231 LOG.error("{}: Failed to register mount point: {}", id, throwable);
236 private DOMRpcService getDOMRpcService(final ActorRef masterReference) {
237 return new ProxyDOMRpcService(setup.getActorSystem(), masterReference, id, actorResponseWaitTime);
240 private CheckedFuture<SchemaContext, SchemaResolutionException> getSchemaContext(final ActorRef masterReference) {
242 final RemoteYangTextSourceProvider remoteYangTextSourceProvider =
243 new ProxyYangTextSourceProvider(masterReference, getContext(), actorResponseWaitTime);
244 final RemoteSchemaProvider remoteProvider = new RemoteSchemaProvider(remoteYangTextSourceProvider,
245 getContext().dispatcher());
247 sourceIdentifiers.forEach(sourceId ->
248 schemaRegistry.registerSchemaSource(remoteProvider, PotentialSchemaSource.create(sourceId,
249 YangTextSchemaSource.class, PotentialSchemaSource.Costs.REMOTE_IO.getValue())));
251 final SchemaContextFactory schemaContextFactory
252 = schemaRepository.createSchemaContextFactory(SchemaSourceFilter.ALWAYS_ACCEPT);
254 return schemaContextFactory.createSchemaContext(sourceIdentifiers);