opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RemoteRpcImplementation.java

   1 package org.opendaylight.controller.remote.rpc;
   2
   3 import akka.actor.ActorRef;
   4 import akka.dispatch.OnComplete;
   5 import com.google.common.util.concurrent.ListenableFuture;
   6 import com.google.common.util.concurrent.SettableFuture;
   7 import org.opendaylight.controller.remote.rpc.messages.InvokeRpc;
   8 import org.opendaylight.controller.remote.rpc.messages.RpcResponse;
   9 import org.opendaylight.controller.sal.core.api.RoutedRpcDefaultImplementation;
  10 import org.opendaylight.controller.sal.core.api.RpcImplementation;
  11 import org.opendaylight.controller.xml.codec.XmlUtils;
  12 import org.opendaylight.yangtools.yang.common.QName;
  13 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
  14 import org.opendaylight.yangtools.yang.common.RpcResult;
  15 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  16 import org.opendaylight.yangtools.yang.data.api.CompositeNode;
  17 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  18 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
  19 import org.slf4j.Logger;
  20 import org.slf4j.LoggerFactory;
  21 import scala.concurrent.ExecutionContext;
  22
  23 import java.util.Collections;
  24 import java.util.Set;
  25
  26 import static akka.pattern.Patterns.ask;
  27
  28 public class RemoteRpcImplementation implements RpcImplementation, RoutedRpcDefaultImplementation {
  29     private static final Logger LOG = LoggerFactory.getLogger(RemoteRpcImplementation.class);
  30     private final ActorRef rpcBroker;
  31     private final SchemaContext schemaContext;
  32     private final RemoteRpcProviderConfig config;
  33
  34     public RemoteRpcImplementation(ActorRef rpcBroker, SchemaContext schemaContext, RemoteRpcProviderConfig config) {
  35         this.rpcBroker = rpcBroker;
  36         this.schemaContext = schemaContext;
  37         this.config = config;
  38     }
  39
  40     @Override
  41     public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(QName rpc,
  42             YangInstanceIdentifier identifier, CompositeNode input) {
  43         InvokeRpc rpcMsg = new InvokeRpc(rpc, identifier, input);
  44
  45         return executeMsg(rpcMsg);
  46     }
  47
  48     @Override
  49     public Set<QName> getSupportedRpcs() {
  50         // TODO : check if we need to get this from routing registry
  51         return Collections.emptySet();
  52     }
  53
  54     @Override
  55     public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(QName rpc, CompositeNode input) {
  56         InvokeRpc rpcMsg = new InvokeRpc(rpc, null, input);
  57         return executeMsg(rpcMsg);
  58     }
  59
  60     private ListenableFuture<RpcResult<CompositeNode>> executeMsg(InvokeRpc rpcMsg) {
  61
  62         final SettableFuture<RpcResult<CompositeNode>> listenableFuture = SettableFuture.create();
  63
  64         scala.concurrent.Future<Object> future = ask(rpcBroker, rpcMsg, config.getAskDuration());
  65
  66         OnComplete<Object> onComplete = new OnComplete<Object>() {
  67             @Override
  68             public void onComplete(Throwable failure, Object reply) throws Throwable {
  69                 if(failure != null) {
  70                     LOG.error("InvokeRpc failed", failure);
  71
  72                     RpcResult<CompositeNode> rpcResult;
  73                     if(failure instanceof RpcErrorsException) {
  74                         rpcResult = RpcResultBuilder.<CompositeNode>failed().withRpcErrors(
  75                                 ((RpcErrorsException)failure).getRpcErrors()).build();
  76                     } else {
  77                         rpcResult = RpcResultBuilder.<CompositeNode>failed().withError(
  78                                 ErrorType.RPC, failure.getMessage(), failure).build();
  79                     }
  80
  81                     listenableFuture.set(rpcResult);
  82                     return;
  83                 }
  84
  85                 RpcResponse rpcReply = (RpcResponse)reply;
  86                 CompositeNode result = XmlUtils.xmlToCompositeNode(rpcReply.getResultCompositeNode());
  87                 listenableFuture.set(RpcResultBuilder.success(result).build());
  88             }
  89         };
  90
  91         future.onComplete(onComplete, ExecutionContext.Implicits$.MODULE$.global());
  92
  93         return listenableFuture;
  94     }
  95 }