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=d384144f4fde6389861597777bd795b30478e773;hb=168591eb3dfc298b631e8dc51e25b0331c105cfc;hpb=4a8d4efda0828bc0c147dee3644c51baa6ff5a15 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..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,86 +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.ListenableFuture; -import org.opendaylight.controller.remote.rpc.messages.ErrorResponse; -import org.opendaylight.controller.remote.rpc.messages.InvokeRoutedRpc; -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.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 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.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; -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); - } +public class RemoteRpcImplementation implements DOMRpcImplementation { + private static final Logger LOG = LoggerFactory.getLogger(RemoteRpcImplementation.class); - private ListenableFuture> executeMsg(Object rpcMsg) { - ListenableFuture> listenableFuture = null; + private final ActorRef rpcRegistry; + private final RemoteRpcProviderConfig config; - 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()); + public RemoteRpcImplementation(final ActorRef rpcRegistry, final RemoteRpcProviderConfig config) { + this.config = config; + this.rpcRegistry = rpcRegistry; + } - final RpcResultBuilder failed = RpcResultBuilder.failed(); - failed.withError(null, null, e.getMessage(), null, null, e.getCause()); - listenableFuture = Futures.immediateFuture(failed.build()); + @Override + 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() { + + @Override + 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())); + } + } + } + }, ExecutionContext.Implicits$.MODULE$.global()); + return frontEndFuture; } - return listenableFuture; - } + @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()); + } }