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=2886fd987918a99cbe90b5144914439d9d1dc1be;hp=a6fdfd3ff97445a2f28cd48e077d66e22fd24f38;hb=168591eb3dfc298b631e8dc51e25b0331c105cfc;hpb=a54ec60368110d22794602343c934902f6833c65 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 a6fdfd3ff9..2886fd9879 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,96 +1,79 @@ package org.opendaylight.controller.remote.rpc; import static akka.pattern.Patterns.ask; + import akka.actor.ActorRef; import akka.dispatch.OnComplete; +import akka.japi.Pair; 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 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 java.util.List; 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.yangtools.yang.common.RpcError; -import org.opendaylight.yangtools.yang.common.RpcError.ErrorType; -import org.opendaylight.yangtools.yang.common.RpcResultBuilder; +import org.opendaylight.controller.remote.rpc.messages.ExecuteRpc; +import org.opendaylight.controller.remote.rpc.registry.RpcRegistry; +import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.FindRoutersReply; +import org.opendaylight.controller.remote.rpc.utils.LatestEntryRoutingLogic; +import org.opendaylight.controller.sal.connector.api.RpcRouter; import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import scala.concurrent.ExecutionContext; +import scala.concurrent.Future; public class RemoteRpcImplementation implements DOMRpcImplementation { private static final Logger LOG = LoggerFactory.getLogger(RemoteRpcImplementation.class); - private final ActorRef rpcBroker; + + private final ActorRef rpcRegistry; private final RemoteRpcProviderConfig config; - public RemoteRpcImplementation(final ActorRef rpcBroker, final RemoteRpcProviderConfig config) { - this.rpcBroker = rpcBroker; + public RemoteRpcImplementation(final ActorRef rpcRegistry, final RemoteRpcProviderConfig config) { this.config = config; + this.rpcRegistry = rpcRegistry; } @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()); + public CheckedFuture invokeRpc(final DOMRpcIdentifier rpc, + final NormalizedNode input) { + if (input instanceof RemoteRpcInput) { + LOG.warn("Rpc {} was removed during execution or there is loop present. Failing received rpc.", rpc); + return Futures + .immediateFailedCheckedFuture(new DOMRpcImplementationNotAvailableException( + "Rpc implementation for {} was removed during processing.", rpc)); + } + final RemoteDOMRpcFuture frontEndFuture = RemoteDOMRpcFuture.create(rpc.getType().getLastComponent()); + findRouteAsync(rpc).onComplete(new OnComplete() { - 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); + public void onComplete(final Throwable error, final FindRoutersReply routes) throws Throwable { + if (error != null) { + frontEndFuture.failNow(error); + } else { + final List> routePairs = routes.getRouterWithUpdateTime(); + if (routePairs == null || routePairs.isEmpty()) { + frontEndFuture.failNow(new DOMRpcImplementationNotAvailableException( + "No local or remote implementation available for rpc %s", rpc.getType(), error)); + } else { + final ActorRef remoteImplRef = new LatestEntryRoutingLogic(routePairs).select(); + final Object executeRpcMessage = ExecuteRpc.from(rpc, input); + LOG.debug("Found remote actor {} for rpc {} - sending {}", remoteImplRef, rpc.getType(), executeRpcMessage); + frontEndFuture.completeWith(ask(remoteImplRef, executeRpcMessage, config.getAskDuration())); } - - 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 = - NormalizedNodeSerializer.deSerialize(rpcReply.getResultNormalizedNode()); - settableFuture.set(new DefaultDOMRpcResult(result)); } - }; - + }, ExecutionContext.Implicits$.MODULE$.global()); + return frontEndFuture; + } - 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")); - } + @SuppressWarnings({"unchecked", "rawtypes"}) + private Future findRouteAsync(final DOMRpcIdentifier rpc) { + // FIXME: Refactor routeId and message to use DOMRpcIdentifier directly. + final RpcRouter.RouteIdentifier routeId = + new RouteIdentifierImpl(null, rpc.getType().getLastComponent(), rpc.getContextReference()); + final RpcRegistry.Messages.FindRouters findMsg = new RpcRegistry.Messages.FindRouters(routeId); + return (Future) ask(rpcRegistry, findMsg, config.getAskDuration()); } }