X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-remoterpc-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fremote%2Frpc%2FRemoteRpcImplementation.java;h=a7370490dc7b0a1428150cb46a5e8454c34cd25b;hb=refs%2Fchanges%2F20%2F89320%2F2;hp=f80aae4fe2d922c85e8d9dc9841b87ff56e9c495;hpb=ca72e02478f78e4cd0346ee04d5a656da3d20b37;p=controller.git 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 f80aae4fe2..a7370490dc 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,87 +1,38 @@ /* - * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others. All rights reserved. + * Copyright (c) 2014, 2017 Cisco Systems, Inc. and others. All rights reserved. * * This program and the accompanying materials are made available under the * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html */ - 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 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 com.google.common.util.concurrent.ListenableFuture; 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.mdsal.dom.api.DOMRpcIdentifier; +import org.opendaylight.mdsal.dom.api.DOMRpcImplementation; +import org.opendaylight.mdsal.dom.api.DOMRpcResult; 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 rpcRegistry; - private final RemoteRpcProviderConfig config; - public RemoteRpcImplementation(final ActorRef rpcRegistry, final RemoteRpcProviderConfig config) { - this.config = config; - this.rpcRegistry = rpcRegistry; +/** + * A {@link DOMRpcImplementation} which routes invocation requests to a remote invoker actor. + * + * @author Robert Varga + */ +final class RemoteRpcImplementation extends AbstractRemoteImplementation implements DOMRpcImplementation { + RemoteRpcImplementation(final ActorRef remoteInvoker, final RemoteOpsProviderConfig config) { + super(remoteInvoker, config); } @Override - public CheckedFuture invokeRpc(final DOMRpcIdentifier rpc, + public ListenableFuture 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 new RemoteDOMRpcFuture(rpc.getType(), ask(ExecuteRpc.from(rpc, input))); } - @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()); + @Override + public long invocationCost() { + return COST; } }