2 * Copyright (c) 2014 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.controller.remote.rpc;
11 import akka.actor.ActorRef;
12 import akka.actor.Props;
13 import akka.japi.Creator;
14 import org.opendaylight.controller.remote.rpc.messages.ErrorResponse;
15 import org.opendaylight.controller.remote.rpc.messages.ExecuteRpc;
16 import org.opendaylight.controller.remote.rpc.messages.GetRoutedRpc;
17 import org.opendaylight.controller.remote.rpc.messages.GetRoutedRpcReply;
18 import org.opendaylight.controller.remote.rpc.messages.GetRpc;
19 import org.opendaylight.controller.remote.rpc.messages.GetRpcReply;
20 import org.opendaylight.controller.remote.rpc.messages.InvokeRoutedRpc;
21 import org.opendaylight.controller.remote.rpc.messages.InvokeRpc;
22 import org.opendaylight.controller.remote.rpc.messages.RpcResponse;
23 import org.opendaylight.controller.remote.rpc.utils.XmlUtils;
24 import org.opendaylight.controller.sal.core.api.Broker;
25 import org.opendaylight.yangtools.yang.common.RpcResult;
26 import org.opendaylight.yangtools.yang.data.api.CompositeNode;
27 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
28 import org.slf4j.Logger;
29 import org.slf4j.LoggerFactory;
31 import java.util.concurrent.Future;
34 * Actor to initiate execution of remote RPC on other nodes of the cluster.
37 public class RpcBroker extends AbstractUntypedActor {
39 private static final Logger LOG = LoggerFactory.getLogger(RpcBroker.class);
40 private final Broker.ProviderSession brokerSession;
41 private final ActorRef rpcRegistry;
42 private final SchemaContext schemaContext;
44 private RpcBroker(Broker.ProviderSession brokerSession, ActorRef rpcRegistry, SchemaContext schemaContext){
45 this.brokerSession = brokerSession;
46 this.rpcRegistry = rpcRegistry;
47 this.schemaContext = schemaContext;
50 public static Props props(final Broker.ProviderSession brokerSession, final ActorRef rpcRegistry, final SchemaContext schemaContext){
51 return Props.create(new Creator<RpcBroker>(){
54 public RpcBroker create() throws Exception {
55 return new RpcBroker(brokerSession, rpcRegistry, schemaContext);
60 protected void handleReceive(Object message) throws Exception {
61 if(message instanceof InvokeRoutedRpc) {
62 invokeRemoteRoutedRpc((InvokeRoutedRpc) message);
63 } else if(message instanceof InvokeRpc) {
64 invokeRemoteRpc((InvokeRpc) message);
65 } else if(message instanceof ExecuteRpc) {
66 executeRpc((ExecuteRpc) message);
70 private void invokeRemoteRoutedRpc(InvokeRoutedRpc msg) {
71 // Look up the remote actor to execute rpc
72 LOG.debug("Looking up the remote actor for route {}", msg);
74 RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, msg.getRpc(), msg.getIdentifier());
75 GetRoutedRpc routedRpcMsg = new GetRoutedRpc(routeId);
76 GetRoutedRpcReply rpcReply = (GetRoutedRpcReply)ActorUtil.executeLocalOperation(rpcRegistry, routedRpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
78 String remoteActorPath = rpcReply.getRoutePath();
79 if(remoteActorPath == null) {
80 LOG.debug("No remote actor found for rpc execution.");
82 getSender().tell(new ErrorResponse(
83 new IllegalStateException("No remote actor found for rpc execution.")), self());
86 ExecuteRpc executeMsg = new ExecuteRpc(XmlUtils.inputCompositeNodeToXml(msg.getInput(), schemaContext), msg.getRpc());
88 Object operationRes = ActorUtil.executeRemoteOperation(this.context().actorSelection(remoteActorPath),
89 executeMsg, ActorUtil.REMOTE_ASK_DURATION, ActorUtil.REMOTE_AWAIT_DURATION);
91 getSender().tell(operationRes, self());
93 } catch (Exception e) {
94 LOG.error(e.toString());
95 getSender().tell(new ErrorResponse(e), self());
99 private void invokeRemoteRpc(InvokeRpc msg) {
100 // Look up the remote actor to execute rpc
101 LOG.debug("Looking up the remote actor for route {}", msg);
103 RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, msg.getRpc(), null);
104 GetRpc rpcMsg = new GetRpc(routeId);
105 GetRpcReply rpcReply = (GetRpcReply)ActorUtil.executeLocalOperation(rpcRegistry, rpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
106 String remoteActorPath = rpcReply.getRoutePath();
108 if(remoteActorPath == null) {
109 LOG.debug("No remote actor found for rpc {{}}.", msg.getRpc());
111 getSender().tell(new ErrorResponse(
112 new IllegalStateException("No remote actor found for rpc execution of : " + msg.getRpc())), self());
115 ExecuteRpc executeMsg = new ExecuteRpc(XmlUtils.inputCompositeNodeToXml(msg.getInput(), schemaContext), msg.getRpc());
116 Object operationRes = ActorUtil.executeRemoteOperation(this.context().actorSelection(remoteActorPath),
117 executeMsg, ActorUtil.REMOTE_ASK_DURATION, ActorUtil.REMOTE_AWAIT_DURATION);
119 getSender().tell(operationRes, self());
121 } catch (Exception e) {
122 LOG.error(e.toString());
123 getSender().tell(new ErrorResponse(e), self());
127 private void executeRpc(ExecuteRpc msg) {
128 LOG.debug("Executing rpc for rpc {}", msg.getRpc());
130 Future<RpcResult<CompositeNode>> rpc = brokerSession.rpc(msg.getRpc(), XmlUtils.inputXmlToCompositeNode(msg.getRpc(), msg.getInputCompositeNode(), schemaContext));
131 RpcResult<CompositeNode> rpcResult = rpc != null ? rpc.get():null;
132 CompositeNode result = rpcResult != null ? rpcResult.getResult() : null;
133 getSender().tell(new RpcResponse(XmlUtils.outputCompositeNodeToXml(result, schemaContext)), self());
134 } catch (Exception e) {
135 LOG.error(e.toString());
136 getSender().tell(new ErrorResponse(e), self());