X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-remoterpc-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fremote%2Frpc%2FRemoteRpcImplementation.java;h=23d1e85e23c355721c31e22bda7d6e7232c9cfd1;hp=d384144f4fde6389861597777bd795b30478e773;hb=1b37a887301c6b9159ddb76492c4a3a820c0dfe8;hpb=3946a3471905eb59a0882eec5df0edd92d1126f1 diff --git a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RemoteRpcImplementation.java b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RemoteRpcImplementation.java index d384144f4f..23d1e85e23 100644 --- a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RemoteRpcImplementation.java +++ b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/RemoteRpcImplementation.java @@ -1,86 +1,104 @@ package org.opendaylight.controller.remote.rpc; +import static akka.pattern.Patterns.ask; import akka.actor.ActorRef; +import akka.dispatch.OnComplete; +import com.google.common.util.concurrent.CheckedFuture; import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.JdkFutureAdapters; import com.google.common.util.concurrent.ListenableFuture; -import org.opendaylight.controller.remote.rpc.messages.ErrorResponse; -import org.opendaylight.controller.remote.rpc.messages.InvokeRoutedRpc; +import com.google.common.util.concurrent.SettableFuture; +import java.util.Arrays; +import java.util.Collection; +import java.util.concurrent.ExecutionException; +import org.opendaylight.controller.cluster.datastore.node.utils.serialization.NormalizedNodeSerializer; +import org.opendaylight.controller.md.sal.dom.api.DOMRpcException; +import org.opendaylight.controller.md.sal.dom.api.DOMRpcIdentifier; +import org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementation; +import org.opendaylight.controller.md.sal.dom.api.DOMRpcImplementationNotAvailableException; +import org.opendaylight.controller.md.sal.dom.api.DOMRpcResult; +import org.opendaylight.controller.md.sal.dom.spi.DefaultDOMRpcResult; import org.opendaylight.controller.remote.rpc.messages.InvokeRpc; import org.opendaylight.controller.remote.rpc.messages.RpcResponse; -import org.opendaylight.controller.remote.rpc.utils.ActorUtil; -import org.opendaylight.controller.remote.rpc.utils.XmlUtils; -import org.opendaylight.controller.sal.core.api.RoutedRpcDefaultImplementation; -import org.opendaylight.controller.sal.core.api.RpcImplementation; -import org.opendaylight.yangtools.yang.common.QName; -import org.opendaylight.yangtools.yang.common.RpcResult; +import org.opendaylight.yangtools.yang.common.RpcError; +import org.opendaylight.yangtools.yang.common.RpcError.ErrorType; import org.opendaylight.yangtools.yang.common.RpcResultBuilder; -import org.opendaylight.yangtools.yang.data.api.CompositeNode; -import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier; -import org.opendaylight.yangtools.yang.model.api.SchemaContext; +import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import scala.concurrent.ExecutionContext; -import java.util.Collections; -import java.util.Set; - -public class RemoteRpcImplementation implements RpcImplementation, - RoutedRpcDefaultImplementation { - private static final Logger LOG = LoggerFactory.getLogger(RemoteRpcImplementation.class); - private ActorRef rpcBroker; - private SchemaContext schemaContext; - - public RemoteRpcImplementation(ActorRef rpcBroker, SchemaContext schemaContext) { - this.rpcBroker = rpcBroker; - this.schemaContext = schemaContext; - } - - @Override - public ListenableFuture> invokeRpc(QName rpc, YangInstanceIdentifier identifier, CompositeNode input) { - InvokeRoutedRpc rpcMsg = new InvokeRoutedRpc(rpc, identifier, input); - - return executeMsg(rpcMsg); - } - - @Override - public Set getSupportedRpcs() { - // TODO : check if we need to get this from routing registry - return Collections.emptySet(); - } - - @Override - public ListenableFuture> invokeRpc(QName rpc, CompositeNode input) { - InvokeRpc rpcMsg = new InvokeRpc(rpc, input); - return executeMsg(rpcMsg); - } - - private ListenableFuture> executeMsg(Object rpcMsg) { - ListenableFuture> listenableFuture = null; - - try { - Object response = ActorUtil.executeLocalOperation(rpcBroker, rpcMsg, ActorUtil.ASK_DURATION, ActorUtil.AWAIT_DURATION); - if(response instanceof RpcResponse) { - - RpcResponse rpcResponse = (RpcResponse) response; - CompositeNode result = XmlUtils.xmlToCompositeNode(rpcResponse.getResultCompositeNode()); - listenableFuture = Futures.immediateFuture(RpcResultBuilder.success(result).build()); - - } else if(response instanceof ErrorResponse) { - - ErrorResponse errorResponse = (ErrorResponse) response; - Exception e = errorResponse.getException(); - final RpcResultBuilder failed = RpcResultBuilder.failed(); - failed.withError(null, null, e.getMessage(), null, null, e.getCause()); - listenableFuture = Futures.immediateFuture(failed.build()); - - } - } catch (Exception e) { - LOG.error("Error occurred while invoking RPC actor {}", e.toString()); - - final RpcResultBuilder failed = RpcResultBuilder.failed(); - failed.withError(null, null, e.getMessage(), null, null, e.getCause()); - listenableFuture = Futures.immediateFuture(failed.build()); +public class RemoteRpcImplementation implements DOMRpcImplementation { + private static final Logger LOG = LoggerFactory.getLogger(RemoteRpcImplementation.class); + private final ActorRef rpcBroker; + private final RemoteRpcProviderConfig config; + + public RemoteRpcImplementation(final ActorRef rpcBroker, final RemoteRpcProviderConfig config) { + this.rpcBroker = rpcBroker; + this.config = config; } - return listenableFuture; - } + @Override + public CheckedFuture invokeRpc(final DOMRpcIdentifier rpc, final NormalizedNode input) { + final InvokeRpc rpcMsg = new InvokeRpc(rpc.getType().getLastComponent(), rpc.getContextReference(), input); + + final SettableFuture settableFuture = SettableFuture.create(); + + final ListenableFuture listenableFuture = + JdkFutureAdapters.listenInPoolThread(settableFuture); + + final scala.concurrent.Future future = ask(rpcBroker, rpcMsg, config.getAskDuration()); + + final OnComplete onComplete = new OnComplete() { + @Override + public void onComplete(final Throwable failure, final Object reply) throws Throwable { + if(failure != null) { + + // When we return a failure to the caller they can choose to log it if they like + // so here we just do basic warn logging by default and log the stack trace only when debug + // is enabled + + LOG.warn("InvokeRpc failed rpc = {}, identifier = {}", rpcMsg.getRpc(), rpcMsg.getIdentifier()); + + if(LOG.isDebugEnabled()){ + LOG.debug("Detailed Error", failure); + } + + final String message = String.format("Execution of RPC %s failed", rpcMsg.getRpc()); + Collection errors = ((RpcErrorsException)failure).getRpcErrors(); + if(errors == null || errors.size() == 0) { + errors = Arrays.asList(RpcResultBuilder.newError(ErrorType.RPC, null, message)); + } + final DOMRpcResult rpcResult = new DefaultDOMRpcResult(errors); + + settableFuture.set(rpcResult); + return; + } + + final RpcResponse rpcReply = (RpcResponse)reply; + final NormalizedNode result; + + if(rpcReply.getResultNormalizedNode() == null){ + result = null; + LOG.debug("Received response for invoke rpc : {} result is null", rpcMsg.getRpc()); + } else { + result = NormalizedNodeSerializer.deSerialize(rpcReply.getResultNormalizedNode()); + LOG.debug("Received response for invoke rpc : {} result : {}", rpcMsg.getRpc(), result); + } + + settableFuture.set(new DefaultDOMRpcResult(result)); + } + }; + + + future.onComplete(onComplete, ExecutionContext.Implicits$.MODULE$.global()); + // FIXME find non blocking way for implementation + try { + return Futures.immediateCheckedFuture(listenableFuture.get()); + } + catch (InterruptedException | ExecutionException e) { + LOG.debug("Unexpected remote RPC exception.", e); + return Futures.immediateFailedCheckedFuture((DOMRpcException) new DOMRpcImplementationNotAvailableException(e, "Unexpected remote RPC exception")); + } + } }