BUG 3566 : Get remote-rpc working again
[controller.git] / 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
81                 if(rpcReply.getResultNormalizedNode() == null){
82                     result = null;
83                     LOG.debug("Received response for invoke rpc : {} result is null", rpcMsg.getRpc());
84                 } else {
85                     result = NormalizedNodeSerializer.deSerialize(rpcReply.getResultNormalizedNode());
86                     LOG.debug("Received response for invoke rpc : {} result : {}", rpcMsg.getRpc(), result);
87                 }
88
89                 settableFuture.set(new DefaultDOMRpcResult(result));
90             }
91         };
92
93
94         future.onComplete(onComplete, ExecutionContext.Implicits$.MODULE$.global());
95         // FIXME find non blocking way for implementation
96         try {
97             return Futures.immediateCheckedFuture(listenableFuture.get());
98         }
99         catch (InterruptedException | ExecutionException e) {
100             LOG.debug("Unexpected remote RPC exception.", e);
101             return Futures.immediateFailedCheckedFuture((DOMRpcException) new DOMRpcImplementationNotAvailableException(e, "Unexpected remote RPC exception"));
102         }
103     }
104 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.