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                     LOG.error("InvokeRpc failed", failure);
  57
  58                     final String message = String.format("Execution of RPC %s failed",  rpcMsg.getRpc());
  59                     Collection<RpcError> errors = ((RpcErrorsException)failure).getRpcErrors();
  60                     if(errors == null || errors.size() == 0) {
  61                         errors = Arrays.asList(RpcResultBuilder.newError(ErrorType.RPC, null, message));
  62                     }
  63                     final DOMRpcResult rpcResult = new DefaultDOMRpcResult(errors);
  64
  65                     settableFuture.set(rpcResult);
  66                     return;
  67                 }
  68
  69                 final RpcResponse rpcReply = (RpcResponse)reply;
  70                 final NormalizedNode<?, ?> result =
  71                         NormalizedNodeSerializer.deSerialize(rpcReply.getResultNormalizedNode());
  72                 settableFuture.set(new DefaultDOMRpcResult(result));
  73             }
  74         };
  75
  76
  77         future.onComplete(onComplete, ExecutionContext.Implicits$.MODULE$.global());
  78         // FIXME find non blocking way for implementation
  79         try {
  80             return Futures.immediateCheckedFuture(listenableFuture.get());
  81         }
  82         catch (InterruptedException | ExecutionException e) {
  83             LOG.debug("Unexpected remote RPC exception.", e);
  84             return Futures.immediateFailedCheckedFuture((DOMRpcException) new DOMRpcImplementationNotAvailableException(e, "Unexpected remote RPC exception"));
  85         }
  86     }
  87 }