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 because of %s",
  68                             rpcMsg.getRpc(), failure.getMessage());
  69                     Collection<RpcError> errors = Arrays.asList(RpcResultBuilder.newError(ErrorType.RPC, null, message));
  70                     if(failure instanceof RpcErrorsException) {
  71                         errors = ((RpcErrorsException) failure).getRpcErrors();
  72                         if (errors == null || errors.size() == 0) {
  73                             errors = Arrays.asList(RpcResultBuilder.newError(ErrorType.RPC, null, message));
  74                         }
  75                     }
  76                     final DOMRpcResult rpcResult = new DefaultDOMRpcResult(errors);
  77
  78                     settableFuture.set(rpcResult);
  79                     return;
  80                 }
  81
  82                 final RpcResponse rpcReply = (RpcResponse)reply;
  83                 final NormalizedNode<?, ?> result;
  84
  85                 if(rpcReply.getResultNormalizedNode() == null){
  86                     result = null;
  87                     LOG.debug("Received response for invoke rpc : {} result is null", rpcMsg.getRpc());
  88                 } else {
  89                     result = NormalizedNodeSerializer.deSerialize(rpcReply.getResultNormalizedNode());
  90                     LOG.debug("Received response for invoke rpc : {} result : {}", rpcMsg.getRpc(), result);
  91                 }
  92
  93                 settableFuture.set(new DefaultDOMRpcResult(result));
  94             }
  95         };
  96
  97
  98         future.onComplete(onComplete, ExecutionContext.Implicits$.MODULE$.global());
  99         // FIXME find non blocking way for implementation
 100         try {
 101             return Futures.immediateCheckedFuture(listenableFuture.get());
 102         }
 103         catch (InterruptedException | ExecutionException e) {
 104             LOG.debug("Unexpected remote RPC exception.", e);
 105             return Futures.immediateFailedCheckedFuture((DOMRpcException) new DOMRpcImplementationNotAvailableException(e, "Unexpected remote RPC exception"));
 106         }
 107     }
 108 }