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 static akka.pattern.Patterns.ask;
   4 import akka.actor.ActorRef;
   5 import akka.dispatch.OnComplete;
   6 import com.google.common.util.concurrent.CheckedFuture;
   7 import com.google.common.util.concurrent.Futures;
   8 import com.google.common.util.concurrent.JdkFutureAdapters;
   9 import com.google.common.util.concurrent.ListenableFuture;
  10 import com.google.common.util.concurrent.SettableFuture;
  11 import java.util.Arrays;
  12 import java.util.Collection;
  13 import java.util.concurrent.ExecutionException;
  14 import org.opendaylight.controller.cluster.datastore.node.utils.serialization.NormalizedNodeSerializer;
  15 import org.opendaylight.controller.md.sal.dom.api.DOMRpcException;
  16 import org.opendaylight.controller.md.sal.dom.api.DOMRpcIdentifier;
  17 import org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementation;
  18 import org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementationNotAvailableException;
  19 import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult;
  20 import org.opendaylight.controller.md.sal.dom.spi.DefaultDOMRpcResult;
  21 import org.opendaylight.controller.remote.rpc.messages.InvokeRpc;
  22 import org.opendaylight.controller.remote.rpc.messages.RpcResponse;
  23 import org.opendaylight.yangtools.yang.common.RpcError;
  24 import org.opendaylight.yangtools.yang.common.RpcError.ErrorType;
  25 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
  26 import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29 import scala.concurrent.ExecutionContext;
  30
  31 public class RemoteRpcImplementation implements DOMRpcImplementation {
  32     private static final Logger LOG = LoggerFactory.getLogger(RemoteRpcImplementation.class);
  33     private final ActorRef rpcBroker;
  34     private final RemoteRpcProviderConfig config;
  35
  36     public RemoteRpcImplementation(final ActorRef rpcBroker, final RemoteRpcProviderConfig config) {
  37         this.rpcBroker = rpcBroker;
  38         this.config = config;
  39     }
  40
  41     @Override
  42     public CheckedFuture<DOMRpcResult, DOMRpcException> invokeRpc(final DOMRpcIdentifier rpc, final NormalizedNode<?, ?> input) {
  43         final InvokeRpc rpcMsg = new InvokeRpc(rpc.getType().getLastComponent(), rpc.getContextReference(), input);
  44
  45         final SettableFuture<DOMRpcResult> settableFuture = SettableFuture.create();
  46
  47         final ListenableFuture<DOMRpcResult> listenableFuture =
  48                 JdkFutureAdapters.listenInPoolThread(settableFuture);
  49
  50         final scala.concurrent.Future<Object> future = ask(rpcBroker, rpcMsg, config.getAskDuration());
  51
  52         final OnComplete<Object> onComplete = new OnComplete<Object>() {
  53             @Override
  54             public void onComplete(final Throwable failure, final Object reply) throws Throwable {
  55                 if(failure != null) {
  56
  57                     // When we return a failure to the caller they can choose to log it if they like
  58                     // so here we just do basic warn logging by default and log the stack trace only when debug
  59                     // is enabled
  60
  61                     LOG.warn("InvokeRpc failed rpc = {}, identifier = {}", rpcMsg.getRpc(), rpcMsg.getIdentifier());
  62
  63                     if(LOG.isDebugEnabled()){
  64                         LOG.debug("Detailed Error", failure);
  65                     }
  66
  67                     final String message = String.format("Execution of RPC %s failed",  rpcMsg.getRpc());
  68                     Collection<RpcError> errors = ((RpcErrorsException)failure).getRpcErrors();
  69                     if(errors == null || errors.size() == 0) {
  70                         errors = Arrays.asList(RpcResultBuilder.newError(ErrorType.RPC, null, message));
  71                     }
  72                     final DOMRpcResult rpcResult = new DefaultDOMRpcResult(errors);
  73
  74                     settableFuture.set(rpcResult);
  75                     return;
  76                 }
  77
  78                 final RpcResponse rpcReply = (RpcResponse)reply;
  79                 final NormalizedNode<?, ?> result =
  80                         NormalizedNodeSerializer.deSerialize(rpcReply.getResultNormalizedNode());
  81                 settableFuture.set(new DefaultDOMRpcResult(result));
  82             }
  83         };
  84
  85
  86         future.onComplete(onComplete, ExecutionContext.Implicits$.MODULE$.global());
  87         // FIXME find non blocking way for implementation
  88         try {
  89             return Futures.immediateCheckedFuture(listenableFuture.get());
  90         }
  91         catch (InterruptedException | ExecutionException e) {
  92             LOG.debug("Unexpected remote RPC exception.", e);
  93             return Futures.immediateFailedCheckedFuture((DOMRpcException) new DOMRpcImplementationNotAvailableException(e, "Unexpected remote RPC exception"));
  94         }
  95     }
  96 }