</execution>
</executions>
</plugin>
+
+
+
</plugins>
</build>
<scm>
public static final String RPC_BROKER_PATH= "/user/rpc/rpc-broker";
public static final String RPC_REGISTRY_PATH = "/user/rpc/rpc-registry";
+ public static final String RPC_MANAGER_PATH = "/user/rpc";
}
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;
@Override
public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(QName rpc, YangInstanceIdentifier identifier, CompositeNode input) {
- InvokeRoutedRpc rpcMsg = new InvokeRoutedRpc(rpc, identifier, input);
+ InvokeRpc rpcMsg = new InvokeRpc(rpc, identifier, input);
return executeMsg(rpcMsg);
}
@Override
public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(QName rpc, CompositeNode input) {
- InvokeRpc rpcMsg = new InvokeRpc(rpc, input);
+ InvokeRpc rpcMsg = new InvokeRpc(rpc, null, input);
return executeMsg(rpcMsg);
}
ListenableFuture<RpcResult<CompositeNode>> listenableFuture = null;
try {
- Object response = ActorUtil.executeLocalOperation(rpcBroker, rpcMsg, ActorUtil.ASK_DURATION, ActorUtil.AWAIT_DURATION);
+ Object response = ActorUtil.executeOperation(rpcBroker, rpcMsg, ActorUtil.ASK_DURATION, ActorUtil.AWAIT_DURATION);
if(response instanceof RpcResponse) {
RpcResponse rpcResponse = (RpcResponse) response;
}
} catch (Exception e) {
- LOG.error("Error occurred while invoking RPC actor {}", e.toString());
+ LOG.error("Error occurred while invoking RPC actor {}", e);
final RpcResultBuilder<CompositeNode> failed = RpcResultBuilder.failed();
failed.withError(null, null, e.getMessage(), null, null, e.getCause());
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import org.opendaylight.controller.remote.rpc.messages.UpdateSchemaContext;
-import org.opendaylight.controller.remote.rpc.registry.ClusterWrapper;
-import org.opendaylight.controller.remote.rpc.registry.ClusterWrapperImpl;
import org.opendaylight.controller.sal.core.api.Broker;
import org.opendaylight.controller.sal.core.api.Provider;
import org.opendaylight.controller.sal.core.api.RpcProvisionRegistry;
private void start() {
LOG.info("Starting all rpc listeners and actors.");
// Create actor to handle and sync routing table in cluster
- ClusterWrapper clusterWrapper = new ClusterWrapperImpl(actorSystem);
SchemaService schemaService = brokerSession.getService(SchemaService.class);
schemaContext = schemaService.getGlobalContext();
- rpcManager = actorSystem.actorOf(RpcManager.props(clusterWrapper, schemaContext, brokerSession, rpcProvisionRegistry), ActorConstants.RPC_MANAGER);
+ rpcManager = actorSystem.actorOf(RpcManager.props(schemaContext, brokerSession, rpcProvisionRegistry), ActorConstants.RPC_MANAGER);
LOG.debug("Rpc actors are created.");
}
import com.google.common.base.Preconditions;
import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
-import org.opendaylight.controller.remote.rpc.messages.AddRoutedRpc;
-import org.opendaylight.controller.remote.rpc.messages.RemoveRoutedRpc;
-import org.opendaylight.controller.remote.rpc.utils.ActorUtil;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry;
import org.opendaylight.controller.sal.connector.api.RpcRouter;
import org.opendaylight.controller.sal.core.api.RpcRoutingContext;
import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
public class RoutedRpcListener implements RouteChangeListener<RpcRoutingContext, YangInstanceIdentifier>{
private static final Logger LOG = LoggerFactory.getLogger(RoutedRpcListener.class);
private final ActorRef rpcRegistry;
- private final String actorPath;
- public RoutedRpcListener(ActorRef rpcRegistry, String actorPath) {
+ public RoutedRpcListener(ActorRef rpcRegistry) {
Preconditions.checkNotNull(rpcRegistry, "rpc registry actor should not be null");
- Preconditions.checkNotNull(actorPath, "actor path of rpc broker on current node should not be null");
this.rpcRegistry = rpcRegistry;
- this.actorPath = actorPath;
}
@Override
public void onRouteChange(RouteChange<RpcRoutingContext, YangInstanceIdentifier> routeChange) {
Map<RpcRoutingContext, Set<YangInstanceIdentifier>> announcements = routeChange.getAnnouncements();
- announce(getRouteIdentifiers(announcements));
+ if(announcements != null && announcements.size() > 0){
+ announce(getRouteIdentifiers(announcements));
+ }
Map<RpcRoutingContext, Set<YangInstanceIdentifier>> removals = routeChange.getRemovals();
- remove(getRouteIdentifiers(removals));
+ if(removals != null && removals.size() > 0 ) {
+ remove(getRouteIdentifiers(removals));
+ }
}
/**
*/
private void announce(Set<RpcRouter.RouteIdentifier<?, ?, ?>> announcements) {
LOG.debug("Announcing [{}]", announcements);
- AddRoutedRpc addRpcMsg = new AddRoutedRpc(announcements, actorPath);
- try {
- ActorUtil.executeLocalOperation(rpcRegistry, addRpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
- } catch (Exception e) {
- // Just logging it because Akka API throws this exception
- LOG.error(e.toString());
- }
+ RpcRegistry.Messages.AddOrUpdateRoutes addRpcMsg = new RpcRegistry.Messages.AddOrUpdateRoutes(new ArrayList<>(announcements));
+ rpcRegistry.tell(addRpcMsg, ActorRef.noSender());
}
/**
*/
private void remove(Set<RpcRouter.RouteIdentifier<?, ?, ?>> removals){
LOG.debug("Removing [{}]", removals);
- RemoveRoutedRpc removeRpcMsg = new RemoveRoutedRpc(removals, actorPath);
- try {
- ActorUtil.executeLocalOperation(rpcRegistry, removeRpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
- } catch (Exception e) {
- // Just logging it because Akka API throws this exception
- LOG.error(e.toString());
- }
+ RpcRegistry.Messages.RemoveRoutes removeRpcMsg = new RpcRegistry.Messages.RemoveRoutes(new ArrayList<>(removals));
+ rpcRegistry.tell(removeRpcMsg, ActorRef.noSender());
}
/**
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.japi.Creator;
+import akka.japi.Pair;
import org.opendaylight.controller.remote.rpc.messages.ErrorResponse;
import org.opendaylight.controller.remote.rpc.messages.ExecuteRpc;
-import org.opendaylight.controller.remote.rpc.messages.GetRoutedRpc;
-import org.opendaylight.controller.remote.rpc.messages.GetRoutedRpcReply;
-import org.opendaylight.controller.remote.rpc.messages.GetRpc;
-import org.opendaylight.controller.remote.rpc.messages.GetRpcReply;
-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.LatestEntryRoutingLogic;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry;
import org.opendaylight.controller.remote.rpc.utils.ActorUtil;
+import org.opendaylight.controller.remote.rpc.utils.RoutingLogic;
import org.opendaylight.controller.remote.rpc.utils.XmlUtils;
+import org.opendaylight.controller.sal.connector.api.RpcRouter;
import org.opendaylight.controller.sal.core.api.Broker;
import org.opendaylight.yangtools.yang.common.RpcResult;
import org.opendaylight.yangtools.yang.data.api.CompositeNode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.List;
import java.util.concurrent.Future;
/**
}
@Override
protected void handleReceive(Object message) throws Exception {
- if(message instanceof InvokeRoutedRpc) {
- invokeRemoteRoutedRpc((InvokeRoutedRpc) message);
- } else if(message instanceof InvokeRpc) {
+ if(message instanceof InvokeRpc) {
invokeRemoteRpc((InvokeRpc) message);
} else if(message instanceof ExecuteRpc) {
executeRpc((ExecuteRpc) message);
}
}
- private void invokeRemoteRoutedRpc(InvokeRoutedRpc msg) {
- // Look up the remote actor to execute rpc
- LOG.debug("Looking up the remote actor for route {}", msg);
- try {
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, msg.getRpc(), msg.getIdentifier());
- GetRoutedRpc routedRpcMsg = new GetRoutedRpc(routeId);
- GetRoutedRpcReply rpcReply = (GetRoutedRpcReply) ActorUtil.executeLocalOperation(rpcRegistry, routedRpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
-
- String remoteActorPath = rpcReply.getRoutePath();
- if(remoteActorPath == null) {
- LOG.debug("No remote actor found for rpc execution.");
-
- getSender().tell(new ErrorResponse(
- new IllegalStateException("No remote actor found for rpc execution.")), self());
- } else {
-
- ExecuteRpc executeMsg = new ExecuteRpc(XmlUtils.inputCompositeNodeToXml(msg.getInput(), schemaContext), msg.getRpc());
-
- Object operationRes = ActorUtil.executeRemoteOperation(this.context().actorSelection(remoteActorPath),
- executeMsg, ActorUtil.REMOTE_ASK_DURATION, ActorUtil.REMOTE_AWAIT_DURATION);
-
- getSender().tell(operationRes, self());
- }
- } catch (Exception e) {
- LOG.error(e.toString());
- getSender().tell(new ErrorResponse(e), self());
- }
- }
-
private void invokeRemoteRpc(InvokeRpc msg) {
// Look up the remote actor to execute rpc
LOG.debug("Looking up the remote actor for route {}", msg);
try {
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, msg.getRpc(), null);
- GetRpc rpcMsg = new GetRpc(routeId);
- GetRpcReply rpcReply = (GetRpcReply)ActorUtil.executeLocalOperation(rpcRegistry, rpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
- String remoteActorPath = rpcReply.getRoutePath();
+ // Find router
+ RpcRouter.RouteIdentifier<?,?,?> routeId = new RouteIdentifierImpl(null, msg.getRpc(), msg.getIdentifier());
+ RpcRegistry.Messages.FindRouters rpcMsg = new RpcRegistry.Messages.FindRouters(routeId);
+ RpcRegistry.Messages.FindRoutersReply rpcReply =
+ (RpcRegistry.Messages.FindRoutersReply) ActorUtil.executeOperation(rpcRegistry, rpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
+
+ List<Pair<ActorRef, Long>> actorRefList = rpcReply.getRouterWithUpdateTime();
- if(remoteActorPath == null) {
+ if(actorRefList == null || actorRefList.isEmpty()) {
LOG.debug("No remote actor found for rpc {{}}.", msg.getRpc());
getSender().tell(new ErrorResponse(
- new IllegalStateException("No remote actor found for rpc execution of : " + msg.getRpc())), self());
+ new IllegalStateException("No remote actor found for rpc execution of : " + msg.getRpc())), self());
} else {
+ RoutingLogic logic = new LatestEntryRoutingLogic(actorRefList);
ExecuteRpc executeMsg = new ExecuteRpc(XmlUtils.inputCompositeNodeToXml(msg.getInput(), schemaContext), msg.getRpc());
- Object operationRes = ActorUtil.executeRemoteOperation(this.context().actorSelection(remoteActorPath),
+ Object operationRes = ActorUtil.executeOperation(logic.select(),
executeMsg, ActorUtil.REMOTE_ASK_DURATION, ActorUtil.REMOTE_AWAIT_DURATION);
getSender().tell(operationRes, self());
}
} catch (Exception e) {
- LOG.error(e.toString());
+ LOG.error("invokeRemoteRpc: {}", e);
getSender().tell(new ErrorResponse(e), self());
}
}
+
+
private void executeRpc(ExecuteRpc msg) {
LOG.debug("Executing rpc for rpc {}", msg.getRpc());
try {
- Future<RpcResult<CompositeNode>> rpc = brokerSession.rpc(msg.getRpc(), XmlUtils.inputXmlToCompositeNode(msg.getRpc(), msg.getInputCompositeNode(), schemaContext));
+ Future<RpcResult<CompositeNode>> rpc = brokerSession.rpc(msg.getRpc(),
+ XmlUtils.inputXmlToCompositeNode(msg.getRpc(), msg.getInputCompositeNode(), schemaContext));
RpcResult<CompositeNode> rpcResult = rpc != null ? rpc.get():null;
CompositeNode result = rpcResult != null ? rpcResult.getResult() : null;
getSender().tell(new RpcResponse(XmlUtils.outputCompositeNodeToXml(result, schemaContext)), self());
} catch (Exception e) {
- LOG.error(e.toString());
+ LOG.error("executeRpc: {}", e);
getSender().tell(new ErrorResponse(e), self());
}
}
import akka.actor.ActorRef;
-import org.opendaylight.controller.remote.rpc.messages.AddRpc;
-import org.opendaylight.controller.remote.rpc.messages.RemoveRpc;
-import org.opendaylight.controller.remote.rpc.utils.ActorUtil;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry;
+import org.opendaylight.controller.sal.connector.api.RpcRouter;
import org.opendaylight.controller.sal.core.api.RpcRegistrationListener;
import org.opendaylight.yangtools.yang.common.QName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
+import java.util.List;
+
public class RpcListener implements RpcRegistrationListener{
private static final Logger LOG = LoggerFactory.getLogger(RpcListener.class);
private final ActorRef rpcRegistry;
- private final String actorPath;
- public RpcListener(ActorRef rpcRegistry, String actorPath) {
+ public RpcListener(ActorRef rpcRegistry) {
this.rpcRegistry = rpcRegistry;
- this.actorPath = actorPath;
}
@Override
public void onRpcImplementationAdded(QName rpc) {
LOG.debug("Adding registration for [{}]", rpc);
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, rpc, null);
- AddRpc addRpcMsg = new AddRpc(routeId, actorPath);
- try {
- ActorUtil.executeLocalOperation(rpcRegistry, addRpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
- LOG.debug("Route added [{}-{}]", routeId, this.actorPath);
- } catch (Exception e) {
- // Just logging it because Akka API throws this exception
- LOG.error(e.toString());
- }
-
+ RpcRouter.RouteIdentifier<?,?,?> routeId = new RouteIdentifierImpl(null, rpc, null);
+ List<RpcRouter.RouteIdentifier<?,?,?>> routeIds = new ArrayList<>();
+ routeIds.add(routeId);
+ RpcRegistry.Messages.AddOrUpdateRoutes addRpcMsg = new RpcRegistry.Messages.AddOrUpdateRoutes(routeIds);
+ rpcRegistry.tell(addRpcMsg, ActorRef.noSender());
}
@Override
public void onRpcImplementationRemoved(QName rpc) {
LOG.debug("Removing registration for [{}]", rpc);
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, rpc, null);
- RemoveRpc removeRpcMsg = new RemoveRpc(routeId);
- try {
- ActorUtil.executeLocalOperation(rpcRegistry, removeRpcMsg, ActorUtil.LOCAL_ASK_DURATION, ActorUtil.LOCAL_AWAIT_DURATION);
- } catch (Exception e) {
- // Just logging it because Akka API throws this exception
- LOG.error(e.toString());
- }
+ RpcRouter.RouteIdentifier<?,?,?> routeId = new RouteIdentifierImpl(null, rpc, null);
+ List<RpcRouter.RouteIdentifier<?,?,?>> routeIds = new ArrayList<>();
+ routeIds.add(routeId);
+ RpcRegistry.Messages.RemoveRoutes removeRpcMsg = new RpcRegistry.Messages.RemoveRoutes(routeIds);
+ rpcRegistry.tell(removeRpcMsg, ActorRef.noSender());
}
}
import akka.japi.Creator;
import akka.japi.Function;
import org.opendaylight.controller.remote.rpc.messages.UpdateSchemaContext;
-import org.opendaylight.controller.remote.rpc.registry.ClusterWrapper;
-import org.opendaylight.controller.remote.rpc.registry.RpcRegistryOld;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry;
import org.opendaylight.controller.sal.core.api.Broker;
import org.opendaylight.controller.sal.core.api.RpcProvisionRegistry;
import org.opendaylight.yangtools.yang.common.QName;
private static final Logger LOG = LoggerFactory.getLogger(RpcManager.class);
private SchemaContext schemaContext;
- private final ClusterWrapper clusterWrapper;
private ActorRef rpcBroker;
private ActorRef rpcRegistry;
private final Broker.ProviderSession brokerSession;
private RemoteRpcImplementation rpcImplementation;
private final RpcProvisionRegistry rpcProvisionRegistry;
- private RpcManager(ClusterWrapper clusterWrapper, SchemaContext schemaContext,
+ private RpcManager(SchemaContext schemaContext,
Broker.ProviderSession brokerSession, RpcProvisionRegistry rpcProvisionRegistry) {
- this.clusterWrapper = clusterWrapper;
this.schemaContext = schemaContext;
this.brokerSession = brokerSession;
this.rpcProvisionRegistry = rpcProvisionRegistry;
}
- public static Props props(final ClusterWrapper clusterWrapper, final SchemaContext schemaContext,
+ public static Props props(final SchemaContext schemaContext,
final Broker.ProviderSession brokerSession, final RpcProvisionRegistry rpcProvisionRegistry) {
return Props.create(new Creator<RpcManager>() {
@Override
public RpcManager create() throws Exception {
- return new RpcManager(clusterWrapper, schemaContext, brokerSession, rpcProvisionRegistry);
+ return new RpcManager(schemaContext, brokerSession, rpcProvisionRegistry);
}
});
}
private void createRpcActors() {
LOG.debug("Create rpc registry and broker actors");
- rpcRegistry = getContext().actorOf(RpcRegistryOld.props(clusterWrapper), ActorConstants.RPC_REGISTRY);
+
+ rpcRegistry = getContext().actorOf(Props.create(RpcRegistry.class), ActorConstants.RPC_REGISTRY);
+
rpcBroker = getContext().actorOf(RpcBroker.props(brokerSession, rpcRegistry, schemaContext), ActorConstants.RPC_BROKER);
+ RpcRegistry.Messages.SetLocalRouter localRouter = new RpcRegistry.Messages.SetLocalRouter(rpcBroker);
+ rpcRegistry.tell(localRouter, self());
}
private void startListeners() {
LOG.debug("Registers rpc listeners");
- String rpcBrokerPath = clusterWrapper.getAddress().toString() + ActorConstants.RPC_BROKER_PATH;
- rpcListener = new RpcListener(rpcRegistry, rpcBrokerPath);
- routeChangeListener = new RoutedRpcListener(rpcRegistry, rpcBrokerPath);
+ rpcListener = new RpcListener(rpcRegistry);
+ routeChangeListener = new RoutedRpcListener(rpcRegistry);
rpcImplementation = new RemoteRpcImplementation(rpcBroker, schemaContext);
brokerSession.addRpcRegistrationListener(rpcListener);
+++ /dev/null
-/*
- * Copyright (c) 2014 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.messages;
-
-import com.google.common.base.Preconditions;
-import org.opendaylight.controller.sal.connector.api.RpcRouter;
-
-import java.io.Serializable;
-import java.util.Set;
-
-public class AddRoutedRpc implements Serializable {
-
- private final Set<RpcRouter.RouteIdentifier<?, ?, ?>> announcements;
- private final String actorPath;
-
- public AddRoutedRpc(final Set<RpcRouter.RouteIdentifier<?, ?, ?>> announcements, final String actorPath) {
- Preconditions.checkNotNull(announcements, "Route identifier should not be null");
- Preconditions.checkNotNull(actorPath, "Actor path should not be null");
-
- this.announcements = announcements;
- this.actorPath = actorPath;
- }
-
- public Set<RpcRouter.RouteIdentifier<?, ?, ?>> getAnnouncements() {
- return announcements;
- }
-
- public String getActorPath() {
- return actorPath;
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.messages;
-
-import com.google.common.base.Preconditions;
-import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
-
-import java.io.Serializable;
-
-public class AddRpc implements Serializable {
-
- private final RouteIdentifierImpl routeId;
- private final String actorPath;
-
- public AddRpc(final RouteIdentifierImpl routeId, final String actorPath) {
- Preconditions.checkNotNull(routeId, "Route identifier should not be null");
- Preconditions.checkNotNull(actorPath, "Actor path should not be null");
-
- this.routeId = routeId;
- this.actorPath = actorPath;
- }
-
- public RouteIdentifierImpl getRouteId() {
- return routeId;
- }
-
- public String getActorPath() {
- return actorPath;
- }
-}
\ No newline at end of file
+++ /dev/null
-/*
- * Copyright (c) 2014 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.messages;
-
-
-import com.google.common.base.Preconditions;
-import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
-
-import java.io.Serializable;
-
-public class GetRoutedRpc implements Serializable {
-
- private final RouteIdentifierImpl routeId;
-
- public GetRoutedRpc(final RouteIdentifierImpl routeId) {
- Preconditions.checkNotNull(routeId, "route id should not be null");
- this.routeId = routeId;
- }
-
- public RouteIdentifierImpl getRouteId() {
- return routeId;
- }
-}
+++ /dev/null
-package org.opendaylight.controller.remote.rpc.messages;
-
-/*
- * Copyright (c) 2014 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
- */
-
-import java.io.Serializable;
-
-public class GetRoutedRpcReply implements Serializable {
-
- private final String routePath;
-
- public GetRoutedRpcReply(final String routePath) {
- this.routePath = routePath;
- }
-
- public String getRoutePath() {
- return routePath;
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.messages;
-
-import com.google.common.base.Preconditions;
-import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
-
-import java.io.Serializable;
-
-public class GetRpc implements Serializable {
-
- private final RouteIdentifierImpl routeId;
-
- public GetRpc(final RouteIdentifierImpl routeId) {
- Preconditions.checkNotNull(routeId, "Route Id should not be null");
- this.routeId = routeId;
- }
-
- public RouteIdentifierImpl getRouteId() {
- return routeId;
- }
-}
+++ /dev/null
-package org.opendaylight.controller.remote.rpc.messages;
-
-/*
- * Copyright (c) 2014 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
- */
-
-import java.io.Serializable;
-
-public class GetRpcReply implements Serializable {
-
- private final String routePath;
-
- public GetRpcReply(final String routePath) {
- this.routePath = routePath;
- }
-
- public String getRoutePath() {
- return routePath;
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.messages;
-
-import com.google.common.base.Preconditions;
-import org.opendaylight.yangtools.yang.common.QName;
-import org.opendaylight.yangtools.yang.data.api.CompositeNode;
-import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
-
-import java.io.Serializable;
-
-public class InvokeRoutedRpc implements Serializable {
-
- private final QName rpc;
- private final YangInstanceIdentifier identifier;
- private final CompositeNode input;
-
- public InvokeRoutedRpc(final QName rpc, final YangInstanceIdentifier identifier, final CompositeNode input) {
- Preconditions.checkNotNull(rpc, "rpc qname should not be null");
- Preconditions.checkNotNull(identifier, "instance identifier of routed rpc should not be null");
- Preconditions.checkNotNull(input, "rpc input should not be null");
-
- this.rpc = rpc;
- this.identifier = identifier;
- this.input = input;
- }
-
- public QName getRpc() {
- return rpc;
- }
-
- public YangInstanceIdentifier getIdentifier() {
- return identifier;
- }
-
- public CompositeNode getInput() {
- return input;
- }
-}
import com.google.common.base.Preconditions;
import org.opendaylight.yangtools.yang.common.QName;
import org.opendaylight.yangtools.yang.data.api.CompositeNode;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import java.io.Serializable;
public class InvokeRpc implements Serializable {
private final QName rpc;
+ private final YangInstanceIdentifier identifier;
private final CompositeNode input;
- public InvokeRpc(final QName rpc, final CompositeNode input) {
+ public InvokeRpc(final QName rpc, final YangInstanceIdentifier identifier, final CompositeNode input) {
Preconditions.checkNotNull(rpc, "rpc qname should not be null");
Preconditions.checkNotNull(input, "rpc input should not be null");
this.rpc = rpc;
+ this.identifier = identifier;
this.input = input;
}
return rpc;
}
+ public YangInstanceIdentifier getIdentifier() {
+ return identifier;
+ }
+
public CompositeNode getInput() {
return input;
}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.messages;
-
-import com.google.common.base.Preconditions;
-import org.opendaylight.controller.sal.connector.api.RpcRouter;
-
-import java.io.Serializable;
-import java.util.Set;
-
-public class RemoveRoutedRpc implements Serializable {
-
- private final Set<RpcRouter.RouteIdentifier<?, ?, ?>> announcements;
- private final String actorPath;
-
- public RemoveRoutedRpc(final Set<RpcRouter.RouteIdentifier<?, ?, ?>> announcements, final String actorPath) {
- Preconditions.checkNotNull(announcements, "Route identifier should not be null");
- Preconditions.checkNotNull(actorPath, "Actor path should not be null");
-
- this.announcements = announcements;
- this.actorPath = actorPath;
- }
-
- public Set<RpcRouter.RouteIdentifier<?, ?, ?>> getAnnouncements() {
- return announcements;
- }
-
- public String getActorPath() {
- return actorPath;
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.messages;
-
-import com.google.common.base.Preconditions;
-import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
-
-import java.io.Serializable;
-
-public class RemoveRpc implements Serializable {
-
- private final RouteIdentifierImpl routeId;
-
- public RemoveRpc(final RouteIdentifierImpl routeId) {
- Preconditions.checkNotNull(routeId, "Route Id should not be null");
-
- this.routeId = routeId;
- }
-
- public RouteIdentifierImpl getRouteId() {
- return routeId;
- }
-}
\ No newline at end of file
+++ /dev/null
-/*
- * Copyright (c) 2014 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.messages;
-
-import org.opendaylight.controller.sal.connector.api.RpcRouter;
-
-import java.io.Serializable;
-import java.util.LinkedHashSet;
-import java.util.Map;
-
-public class RoutingTableData implements Serializable {
- private final Map<RpcRouter.RouteIdentifier<?, ?, ?>, String> rpcMap;
- private final Map<RpcRouter.RouteIdentifier<?, ?, ?>, LinkedHashSet<String>> routedRpcMap;
-
- public RoutingTableData(final Map<RpcRouter.RouteIdentifier<?, ?, ?>, String> rpcMap,
- final Map<RpcRouter.RouteIdentifier<?, ?, ?>, LinkedHashSet<String>> routedRpcMap) {
- this.rpcMap = rpcMap;
- this.routedRpcMap = routedRpcMap;
- }
-
- public Map<RpcRouter.RouteIdentifier<?, ?, ?>, String> getRpcMap() {
- return rpcMap;
- }
-
- public Map<RpcRouter.RouteIdentifier<?, ?, ?>, LinkedHashSet<String>> getRoutedRpcMap() {
- return routedRpcMap;
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.registry;
-
-
-import akka.actor.Address;
-import akka.cluster.ClusterEvent;
-
-public interface ClusterWrapper {
-
- ClusterEvent.CurrentClusterState getState();
-
- Address getAddress();
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.registry;
-
-
-import akka.actor.ActorSystem;
-import akka.actor.Address;
-import akka.cluster.Cluster;
-import akka.cluster.ClusterEvent;
-
-
-public class ClusterWrapperImpl implements ClusterWrapper{
-
- private Cluster cluster;
-
- public ClusterWrapperImpl(ActorSystem actorSystem) {
- cluster = Cluster.get(actorSystem);
- }
-
- @Override
- public ClusterEvent.CurrentClusterState getState() {
- return cluster.state();
- }
-
- @Override
- public Address getAddress() {
- return cluster.selfAddress();
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.registry;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableSet;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.LinkedHashSet;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-
-public class RoutingTableOld<I, R> {
-
- private final Logger LOG = LoggerFactory.getLogger(RoutingTableOld.class);
-
- private ConcurrentMap<I,R> globalRpcMap = new ConcurrentHashMap<>();
- private ConcurrentMap<I, LinkedHashSet<R>> routedRpcMap = new ConcurrentHashMap<>();
-
- public ConcurrentMap<I, R> getGlobalRpcMap() {
- return globalRpcMap;
- }
-
- public ConcurrentMap<I, LinkedHashSet<R>> getRoutedRpcMap() {
- return routedRpcMap;
- }
-
- public R getGlobalRoute(final I routeId) {
- Preconditions.checkNotNull(routeId, "getGlobalRoute: routeId cannot be null!");
- return globalRpcMap.get(routeId);
- }
-
- public void addGlobalRoute(final I routeId, final R route) {
- Preconditions.checkNotNull(routeId, "addGlobalRoute: routeId cannot be null!");
- Preconditions.checkNotNull(route, "addGlobalRoute: route cannot be null!");
- LOG.debug("addGlobalRoute: adding a new route with id[{}] and value [{}]", routeId, route);
- if(globalRpcMap.putIfAbsent(routeId, route) != null) {
- LOG.debug("A route already exist for route id [{}] ", routeId);
- }
- }
-
- public void removeGlobalRoute(final I routeId) {
- Preconditions.checkNotNull(routeId, "removeGlobalRoute: routeId cannot be null!");
- LOG.debug("removeGlobalRoute: removing a new route with id [{}]", routeId);
- globalRpcMap.remove(routeId);
- }
-
- public Set<R> getRoutedRpc(final I routeId) {
- Preconditions.checkNotNull(routeId, "getRoutes: routeId cannot be null!");
- Set<R> routes = routedRpcMap.get(routeId);
-
- if (routes == null) {
- return Collections.emptySet();
- }
-
- return ImmutableSet.copyOf(routes);
- }
-
- public R getLastAddedRoutedRpc(final I routeId) {
-
- Set<R> routes = getRoutedRpc(routeId);
-
- if (routes.isEmpty()) {
- return null;
- }
-
- R route = null;
- Iterator<R> iter = routes.iterator();
- while (iter.hasNext()) {
- route = iter.next();
- }
-
- return route;
- }
-
- public void addRoutedRpc(final I routeId, final R route) {
- Preconditions.checkNotNull(routeId, "addRoute: routeId cannot be null");
- Preconditions.checkNotNull(route, "addRoute: route cannot be null");
- LOG.debug("addRoute: adding a route with k/v [{}/{}]", routeId, route);
- threadSafeAdd(routeId, route);
- }
-
- public void addRoutedRpcs(final Set<I> routeIds, final R route) {
- Preconditions.checkNotNull(routeIds, "addRoutes: routeIds must not be null");
- for (I routeId : routeIds){
- addRoutedRpc(routeId, route);
- }
- }
-
- public void removeRoute(final I routeId, final R route) {
- Preconditions.checkNotNull(routeId, "removeRoute: routeId cannot be null!");
- Preconditions.checkNotNull(route, "removeRoute: route cannot be null!");
-
- LinkedHashSet<R> routes = routedRpcMap.get(routeId);
- if (routes == null) {
- return;
- }
- LOG.debug("removeRoute: removing a new route with k/v [{}/{}]", routeId, route);
- threadSafeRemove(routeId, route);
- }
-
- public void removeRoutes(final Set<I> routeIds, final R route) {
- Preconditions.checkNotNull(routeIds, "removeRoutes: routeIds must not be null");
- for (I routeId : routeIds){
- removeRoute(routeId, route);
- }
- }
-
- /**
- * This method guarantees that no 2 thread over write each other's changes.
- * Just so that we dont end up in infinite loop, it tries for 100 times then throw
- */
- private void threadSafeAdd(final I routeId, final R route) {
-
- for (int i=0;i<100;i++){
-
- LinkedHashSet<R> updatedRoutes = new LinkedHashSet<>();
- updatedRoutes.add(route);
- LinkedHashSet<R> oldRoutes = routedRpcMap.putIfAbsent(routeId, updatedRoutes);
- if (oldRoutes == null) {
- return;
- }
-
- updatedRoutes = new LinkedHashSet<>(oldRoutes);
- updatedRoutes.add(route);
-
- if (routedRpcMap.replace(routeId, oldRoutes, updatedRoutes)) {
- return;
- }
- }
- //the method did not already return means it failed to add route in 100 attempts
- throw new IllegalStateException("Failed to add route [" + routeId + "]");
- }
-
- /**
- * This method guarantees that no 2 thread over write each other's changes.
- * Just so that we dont end up in infinite loop, it tries for 100 times then throw
- */
- private void threadSafeRemove(final I routeId, final R route) {
- LinkedHashSet<R> updatedRoutes = null;
- for (int i=0;i<100;i++){
- LinkedHashSet<R> oldRoutes = routedRpcMap.get(routeId);
-
- // if route to be deleted is the only entry in the set then remove routeId from the cache
- if ((oldRoutes.size() == 1) && oldRoutes.contains(route)){
- routedRpcMap.remove(routeId);
- return;
- }
-
- // if there are multiple routes for this routeId, remove the route to be deleted only from the set.
- updatedRoutes = new LinkedHashSet<>(oldRoutes);
- updatedRoutes.remove(route);
- if (routedRpcMap.replace(routeId, oldRoutes, updatedRoutes)) {
- return;
- }
-
- }
- //the method did not already return means it failed to remove route in 100 attempts
- throw new IllegalStateException("Failed to remove route [" + routeId + "]");
- }
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.registry;
-
-import akka.actor.ActorSelection;
-import akka.actor.Address;
-import akka.actor.Props;
-import akka.cluster.ClusterEvent;
-import akka.cluster.Member;
-import akka.japi.Creator;
-import org.opendaylight.controller.remote.rpc.AbstractUntypedActor;
-import org.opendaylight.controller.remote.rpc.ActorConstants;
-import org.opendaylight.controller.remote.rpc.messages.AddRoutedRpc;
-import org.opendaylight.controller.remote.rpc.messages.AddRpc;
-import org.opendaylight.controller.remote.rpc.messages.GetRoutedRpc;
-import org.opendaylight.controller.remote.rpc.messages.GetRoutedRpcReply;
-import org.opendaylight.controller.remote.rpc.messages.GetRpc;
-import org.opendaylight.controller.remote.rpc.messages.GetRpcReply;
-import org.opendaylight.controller.remote.rpc.messages.RemoveRoutedRpc;
-import org.opendaylight.controller.remote.rpc.messages.RemoveRpc;
-import org.opendaylight.controller.remote.rpc.messages.RoutingTableData;
-import org.opendaylight.controller.sal.connector.api.RpcRouter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import scala.collection.JavaConversions;
-
-import java.util.LinkedHashSet;
-import java.util.Map;
-import java.util.Random;
-import java.util.Set;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-
-/**
- * This Actor maintains the routing table state and sync it with other nodes in the cluster.
- *
- * A scheduler runs after an interval of time, which pick a random member from the cluster
- * and send the current state of routing table to the member.
- *
- * when a message of routing table data is received, it gets merged with the local routing table
- * to keep the latest data.
- */
-
-public class RpcRegistryOld extends AbstractUntypedActor {
-
- private static final Logger LOG = LoggerFactory.getLogger(RpcRegistryOld.class);
- private RoutingTableOld<RpcRouter.RouteIdentifier<?, ?, ?>, String> routingTable;
- private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
- private final ClusterWrapper clusterWrapper;
- private final ScheduledFuture<?> syncScheduler;
-
- private RpcRegistryOld(ClusterWrapper clusterWrapper){
- this.routingTable = new RoutingTableOld<>();
- this.clusterWrapper = clusterWrapper;
- this.syncScheduler = scheduler.scheduleAtFixedRate(new SendRoutingTable(), 10, 10, TimeUnit.SECONDS);
- }
-
- public static Props props(final ClusterWrapper clusterWrapper){
- return Props.create(new Creator<RpcRegistryOld>(){
-
- @Override
- public RpcRegistryOld create() throws Exception {
- return new RpcRegistryOld(clusterWrapper);
- }
- });
- }
-
- @Override
- protected void handleReceive(Object message) throws Exception {
- LOG.debug("Received message {}", message);
- if(message instanceof RoutingTableData) {
- syncRoutingTable((RoutingTableData) message);
- } else if(message instanceof GetRoutedRpc) {
- getRoutedRpc((GetRoutedRpc) message);
- } else if(message instanceof GetRpc) {
- getRpc((GetRpc) message);
- } else if(message instanceof AddRpc) {
- addRpc((AddRpc) message);
- } else if(message instanceof RemoveRpc) {
- removeRpc((RemoveRpc) message);
- } else if(message instanceof AddRoutedRpc) {
- addRoutedRpc((AddRoutedRpc) message);
- } else if(message instanceof RemoveRoutedRpc) {
- removeRoutedRpc((RemoveRoutedRpc) message);
- }
- }
-
- private void getRoutedRpc(GetRoutedRpc rpcMsg){
- LOG.debug("Get latest routed Rpc location from routing table {}", rpcMsg);
- String remoteActorPath = routingTable.getLastAddedRoutedRpc(rpcMsg.getRouteId());
- GetRoutedRpcReply routedRpcReply = new GetRoutedRpcReply(remoteActorPath);
-
- getSender().tell(routedRpcReply, self());
- }
-
- private void getRpc(GetRpc rpcMsg) {
- LOG.debug("Get global Rpc location from routing table {}", rpcMsg);
- String remoteActorPath = routingTable.getGlobalRoute(rpcMsg.getRouteId());
- GetRpcReply rpcReply = new GetRpcReply(remoteActorPath);
-
- getSender().tell(rpcReply, self());
- }
-
- private void addRpc(AddRpc rpcMsg) {
- LOG.debug("Add Rpc to routing table {}", rpcMsg);
- routingTable.addGlobalRoute(rpcMsg.getRouteId(), rpcMsg.getActorPath());
-
- getSender().tell("Success", self());
- }
-
- private void removeRpc(RemoveRpc rpcMsg) {
- LOG.debug("Removing Rpc to routing table {}", rpcMsg);
- routingTable.removeGlobalRoute(rpcMsg.getRouteId());
-
- getSender().tell("Success", self());
- }
-
- private void addRoutedRpc(AddRoutedRpc rpcMsg) {
- routingTable.addRoutedRpcs(rpcMsg.getAnnouncements(), rpcMsg.getActorPath());
- getSender().tell("Success", self());
- }
-
- private void removeRoutedRpc(RemoveRoutedRpc rpcMsg) {
- routingTable.removeRoutes(rpcMsg.getAnnouncements(), rpcMsg.getActorPath());
- getSender().tell("Success", self());
- }
-
- private void syncRoutingTable(RoutingTableData routingTableData) {
- LOG.debug("Syncing routing table {}", routingTableData);
-
- Map<RpcRouter.RouteIdentifier<?, ?, ?>, String> newRpcMap = routingTableData.getRpcMap();
- Set<RpcRouter.RouteIdentifier<?, ?, ?>> routeIds = newRpcMap.keySet();
- for(RpcRouter.RouteIdentifier<?, ?, ?> routeId : routeIds) {
- routingTable.addGlobalRoute(routeId, newRpcMap.get(routeId));
- }
-
- Map<RpcRouter.RouteIdentifier<?, ?, ?>, LinkedHashSet<String>> newRoutedRpcMap =
- routingTableData.getRoutedRpcMap();
- routeIds = newRoutedRpcMap.keySet();
-
- for(RpcRouter.RouteIdentifier<?, ?, ?> routeId : routeIds) {
- Set<String> routeAddresses = newRoutedRpcMap.get(routeId);
- for(String routeAddress : routeAddresses) {
- routingTable.addRoutedRpc(routeId, routeAddress);
- }
- }
- }
-
- private ActorSelection getRandomRegistryActor() {
- ClusterEvent.CurrentClusterState clusterState = clusterWrapper.getState();
- ActorSelection actor = null;
- Set<Member> members = JavaConversions.asJavaSet(clusterState.members());
- int memberSize = members.size();
- // Don't select yourself
- if(memberSize > 1) {
- Address currentNodeAddress = clusterWrapper.getAddress();
- int index = new Random().nextInt(memberSize);
- int i = 0;
- // keeping previous member, in case when random index member is same as current actor
- // and current actor member is last in set
- Member previousMember = null;
- for(Member member : members){
- if(i == index-1) {
- previousMember = member;
- }
- if(i == index) {
- if(!currentNodeAddress.equals(member.address())) {
- actor = this.context().actorSelection(member.address() + ActorConstants.RPC_REGISTRY_PATH);
- break;
- } else if(index < memberSize-1){ // pick the next element in the set
- index++;
- }
- }
- i++;
- }
- if(actor == null && previousMember != null) {
- actor = this.context().actorSelection(previousMember.address() + ActorConstants.RPC_REGISTRY_PATH);
- }
- }
- return actor;
- }
-
- private class SendRoutingTable implements Runnable {
-
- @Override
- public void run() {
- RoutingTableData routingTableData =
- new RoutingTableData(routingTable.getGlobalRpcMap(), routingTable.getRoutedRpcMap());
- LOG.debug("Sending routing table for sync {}", routingTableData);
- ActorSelection actor = getRandomRegistryActor();
- if(actor != null) {
- actor.tell(routingTableData, self());
- }
- }
- }
-}
package org.opendaylight.controller.remote.rpc.utils;
import akka.actor.ActorRef;
-import akka.actor.ActorSelection;
import akka.util.Timeout;
import scala.concurrent.Await;
import scala.concurrent.Future;
* @param awaitDuration
* @return The response of the operation
*/
- public static Object executeLocalOperation(ActorRef actor, Object message,
- FiniteDuration askDuration, FiniteDuration awaitDuration) throws Exception{
+ public static Object executeOperation(ActorRef actor, Object message,
+ FiniteDuration askDuration, FiniteDuration awaitDuration) throws Exception{
Future<Object> future =
ask(actor, message, new Timeout(askDuration));
return Await.result(future, awaitDuration);
}
- /**
- * Execute an operation on a remote actor and wait for it's response
- * @param actor
- * @param message
- * @param askDuration
- * @param awaitDuration
- * @return
- */
- public static Object executeRemoteOperation(ActorSelection actor, Object message,
- FiniteDuration askDuration, FiniteDuration awaitDuration) throws Exception{
- Future<Object> future =
- ask(actor, message, new Timeout(askDuration));
- return Await.result(future, awaitDuration);
- }
}
--- /dev/null
+/*
+ * Copyright (c) 2014 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.utils;
+
+import akka.actor.ActorRef;
+import akka.japi.Pair;
+import com.google.common.base.Preconditions;
+
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.SortedSet;
+import java.util.TreeSet;
+
+/**
+ * This class will return First Entry
+ */
+public class LatestEntryRoutingLogic implements RoutingLogic{
+
+ private SortedSet<Pair<ActorRef, Long>> actorRefSet;
+
+ public LatestEntryRoutingLogic(Collection<Pair<ActorRef, Long>> entries) {
+ Preconditions.checkNotNull(entries, "Entries should not be null");
+ Preconditions.checkArgument(!entries.isEmpty(), "Entries collection should not be empty");
+
+ actorRefSet = new TreeSet<>(new LatestEntryComparator());
+ actorRefSet.addAll(entries);
+ }
+
+ @Override
+ public ActorRef select() {
+ return actorRefSet.last().first();
+ }
+
+
+ private class LatestEntryComparator implements Comparator<Pair<ActorRef, Long>> {
+
+ @Override
+ public int compare(Pair<ActorRef, Long> o1, Pair<ActorRef, Long> o2) {
+ if(o1 == null && o2 == null) {
+ return 0;
+ }
+ if(o1 == null && o2 != null) {
+ return -1;
+ }
+ if(o1 != null && o2 == null) {
+ return 1;
+ }
+
+ return o1.second().compareTo(o2.second());
+
+ }
+
+ }
+}
+
+
--- /dev/null
+/*
+ * Copyright (c) 2014 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.utils;
+
+import akka.actor.ActorRef;
+
+/**
+ * This Interface is added to abstract out the way rpc execution could be
+ * routed, if more than one node in cluster is capable of executing the rpc.
+ *
+ * We can pick node randomly, round robin manner or based on last updated time etc.
+ */
+
+public interface RoutingLogic {
+
+ ActorRef select();
+}
cluster {
seed-nodes = ["akka.tcp://opendaylight-cluster-rpc@127.0.0.1:2551"]
-
auto-down-unreachable-after = 10s
}
}
--- /dev/null
+/*
+ * Copyright (c) 2014 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 akka.actor.ActorSystem;
+import junit.framework.Assert;
+import org.junit.After;
+import org.junit.Test;
+import org.osgi.framework.Bundle;
+import org.osgi.framework.BundleContext;
+
+import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class ActorSystemFactoryTest {
+ ActorSystem system = null;
+
+ @Test
+ public void testActorSystemCreation(){
+ BundleContext context = mock(BundleContext.class);
+ when(context.getBundle()).thenReturn(mock(Bundle.class));
+ ActorSystemFactory.createInstance(context);
+ system = ActorSystemFactory.getInstance();
+ Assert.assertNotNull(system);
+ // Check illegal state exception
+
+ try {
+ ActorSystemFactory.createInstance(context);
+ fail("Illegal State exception should be thrown, while creating actor system second time");
+ } catch (IllegalStateException e) {
+ }
+ }
+
+ @After
+ public void cleanup() throws InterruptedException {
+ if(system != null) {
+ system.shutdown();
+ }
+ }
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2014 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 akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.testkit.JavaTestKit;
+import com.typesafe.config.ConfigFactory;
+import junit.framework.Assert;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.opendaylight.controller.sal.core.api.Broker;
+import org.opendaylight.controller.sal.core.api.RpcProvisionRegistry;
+import org.opendaylight.controller.sal.core.api.model.SchemaService;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import scala.concurrent.Await;
+import scala.concurrent.duration.Duration;
+
+
+import java.util.concurrent.TimeUnit;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class RemoteRpcProviderTest {
+
+ static ActorSystem system;
+
+
+ @BeforeClass
+ public static void setup() throws InterruptedException {
+ system = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("odl-cluster"));
+ }
+
+ @AfterClass
+ public static void teardown() {
+ JavaTestKit.shutdownActorSystem(system);
+ system = null;
+ }
+
+ @Test
+ public void testRemoteRpcProvider() throws Exception {
+ RemoteRpcProvider rpcProvider = new RemoteRpcProvider(system, mock(RpcProvisionRegistry.class));
+ Broker.ProviderSession session = mock(Broker.ProviderSession.class);
+ SchemaService schemaService = mock(SchemaService.class);
+ when(schemaService.getGlobalContext()). thenReturn(mock(SchemaContext.class));
+ when(session.getService(SchemaService.class)).thenReturn(schemaService);
+ rpcProvider.onSessionInitiated(session);
+ ActorRef actorRef = Await.result(system.actorSelection(ActorConstants.RPC_MANAGER_PATH).resolveOne(Duration.create(1, TimeUnit.SECONDS)),
+ Duration.create(2, TimeUnit.SECONDS));
+ Assert.assertTrue(actorRef.path().toString().contains(ActorConstants.RPC_MANAGER_PATH));
+ }
+
+
+
+}
--- /dev/null
+/*
+ * Copyright (c) 2014 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 akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.testkit.JavaTestKit;
+import com.typesafe.config.ConfigFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.opendaylight.controller.md.sal.common.impl.routing.RoutingUtils;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry;
+import org.opendaylight.controller.sal.core.api.RpcRoutingContext;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+public class RouteRpcListenerTest {
+
+ static ActorSystem system;
+
+
+ @BeforeClass
+ public static void setup() throws InterruptedException {
+ system = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("odl-cluster"));
+ }
+
+ @AfterClass
+ public static void teardown() {
+ JavaTestKit.shutdownActorSystem(system);
+ system = null;
+ }
+
+ @Test
+ public void testRouteAdd() throws URISyntaxException, InterruptedException {
+ new JavaTestKit(system) {
+ {
+ // Test announcements
+ JavaTestKit probeReg = new JavaTestKit(system);
+ ActorRef rpcRegistry = probeReg.getRef();
+
+ RoutedRpcListener rpcListener = new RoutedRpcListener(rpcRegistry);
+
+ QName qName = new QName(new URI("actor2"), "actor2");
+ RpcRoutingContext context = RpcRoutingContext.create(qName, qName);
+ YangInstanceIdentifier identifier = YangInstanceIdentifier.create(new YangInstanceIdentifier.NodeIdentifier(qName));
+ rpcListener.onRouteChange(RoutingUtils.announcementChange(context, identifier));
+
+ probeReg.expectMsgClass(RpcRegistry.Messages.AddOrUpdateRoutes.class);
+ }};
+ }
+
+ @Test
+ public void testRouteRemove() throws URISyntaxException, InterruptedException {
+ new JavaTestKit(system) {
+ {
+ // Test announcements
+ JavaTestKit probeReg = new JavaTestKit(system);
+ ActorRef rpcRegistry = probeReg.getRef();
+
+ RoutedRpcListener rpcListener = new RoutedRpcListener(rpcRegistry);
+
+ QName qName = new QName(new URI("actor2"), "actor2");
+ RpcRoutingContext context = RpcRoutingContext.create(qName, qName);
+ YangInstanceIdentifier identifier = YangInstanceIdentifier.create(new YangInstanceIdentifier.NodeIdentifier(qName));
+ rpcListener.onRouteChange(RoutingUtils.removalChange(context, identifier));
+
+ probeReg.expectMsgClass(RpcRegistry.Messages.RemoveRoutes.class);
+ }};
+ }
+}
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
+import akka.japi.Pair;
import akka.testkit.JavaTestKit;
import com.google.common.util.concurrent.Futures;
+import com.typesafe.config.ConfigFactory;
import junit.framework.Assert;
import org.junit.AfterClass;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.mockito.Mockito;
-import org.opendaylight.controller.remote.rpc.messages.AddRoutedRpc;
-import org.opendaylight.controller.remote.rpc.messages.AddRpc;
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.registry.ClusterWrapper;
-import org.opendaylight.controller.remote.rpc.registry.RpcRegistryOld;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry;
import org.opendaylight.controller.sal.common.util.Rpcs;
-import org.opendaylight.controller.sal.connector.api.RpcRouter;
import org.opendaylight.controller.sal.core.api.Broker;
import org.opendaylight.yangtools.yang.common.QName;
import org.opendaylight.yangtools.yang.common.RpcError;
import org.opendaylight.yangtools.yang.data.api.CompositeNode;
import org.opendaylight.yangtools.yang.data.api.ModifyAction;
import org.opendaylight.yangtools.yang.data.api.Node;
-import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
import org.opendaylight.yangtools.yang.data.impl.ImmutableCompositeNode;
import org.opendaylight.yangtools.yang.model.api.SchemaContext;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.HashSet;
-import java.util.Set;
+import java.util.List;
import java.util.concurrent.Future;
import static org.mockito.Mockito.mock;
public class RpcBrokerTest {
- static ActorSystem system;
+ static ActorSystem node1;
+ static ActorSystem node2;
+ private ActorRef rpcBroker1;
+ private JavaTestKit probeReg1;
+ private ActorRef rpcBroker2;
+ private JavaTestKit probeReg2;
+ private Broker.ProviderSession brokerSession;
@BeforeClass
- public static void setup() {
- system = ActorSystem.create();
+ public static void setup() throws InterruptedException {
+ node1 = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("memberA"));
+ node2 = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("memberB"));
}
@AfterClass
public static void teardown() {
- JavaTestKit.shutdownActorSystem(system);
- system = null;
+ JavaTestKit.shutdownActorSystem(node1);
+ JavaTestKit.shutdownActorSystem(node2);
+ node1 = null;
+ node2 = null;
}
+ @Before
+ public void createActor() {
+ brokerSession = Mockito.mock(Broker.ProviderSession.class);
+ SchemaContext schemaContext = mock(SchemaContext.class);
+ probeReg1 = new JavaTestKit(node1);
+ rpcBroker1 = node1.actorOf(RpcBroker.props(brokerSession, probeReg1.getRef(), schemaContext));
+ probeReg2 = new JavaTestKit(node2);
+ rpcBroker2 = node2.actorOf(RpcBroker.props(brokerSession, probeReg2.getRef(), schemaContext));
+
+ }
@Test
- public void testInvokeRpcError() throws URISyntaxException {
- new JavaTestKit(system) {{
- ActorRef rpcRegistry = system.actorOf(RpcRegistryOld.props(Mockito.mock(ClusterWrapper.class)));
- Broker.ProviderSession brokerSession = Mockito.mock(Broker.ProviderSession.class);
- SchemaContext schemaContext = mock(SchemaContext.class);
- ActorRef rpcBroker = system.actorOf(RpcBroker.props(brokerSession, rpcRegistry, schemaContext));
+ public void testInvokeRpcError() throws Exception {
+ new JavaTestKit(node1) {{
QName rpc = new QName(new URI("noactor1"), "noactor1");
CompositeNode input = new ImmutableCompositeNode(QName.create("ns", "2013-12-09", "no child"), new ArrayList<Node<?>>(), ModifyAction.REPLACE);
- InvokeRpc invokeMsg = new InvokeRpc(rpc, input);
- rpcBroker.tell(invokeMsg, getRef());
+
+
+ InvokeRpc invokeMsg = new InvokeRpc(rpc, null, input);
+ rpcBroker1.tell(invokeMsg, getRef());
+ probeReg1.expectMsgClass(RpcRegistry.Messages.FindRouters.class);
+ probeReg1.reply(new RpcRegistry.Messages.FindRoutersReply(new ArrayList<Pair<ActorRef, Long>>()));
Boolean getMsg = new ExpectMsg<Boolean>("ErrorResponse") {
protected Boolean match(Object in) {
}.get(); // this extracts the received message
Assert.assertTrue(getMsg);
+
}};
}
+
/**
* This test method invokes and executes the remote rpc
*/
@Test
public void testInvokeRpc() throws URISyntaxException {
- new JavaTestKit(system) {{
- ActorRef rpcRegistry = system.actorOf(RpcRegistryOld.props(mock(ClusterWrapper.class)));
- Broker.ProviderSession brokerSession = mock(Broker.ProviderSession.class);
- SchemaContext schemaContext = mock(SchemaContext.class);
- ActorRef rpcBroker = system.actorOf(RpcBroker.props(brokerSession, rpcRegistry, schemaContext));
- ActorRef rpcBrokerRemote = system.actorOf(RpcBroker.props(brokerSession, rpcRegistry, schemaContext), "actor1");
- // Add RPC in table
- QName rpc = new QName(new URI("actor1"), "actor1");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, rpc, null);
- final String route = rpcBrokerRemote.path().toString();
- AddRpc rpcMsg = new AddRpc(routeId, route);
- rpcRegistry.tell(rpcMsg, getRef());
- expectMsgEquals(duration("2 second"), "Success");
-
+ new JavaTestKit(node1) {{
+ QName rpc = new QName(new URI("noactor1"), "noactor1");
// invoke rpc
CompositeNode input = new ImmutableCompositeNode(QName.create("ns", "2013-12-09", "child1"), new ArrayList<Node<?>>(), ModifyAction.REPLACE);
- CompositeNode invokeRpcResult = mock(CompositeNode.class);
- Collection<RpcError> errors = new ArrayList<>();
- RpcResult<CompositeNode> result = Rpcs.getRpcResult(true, invokeRpcResult, errors);
- Future<RpcResult<CompositeNode>> rpcResult = Futures.immediateFuture(result);
- when(brokerSession.rpc(rpc, input)).thenReturn(rpcResult);
- InvokeRpc invokeMsg = new InvokeRpc(rpc, input);
- rpcBroker.tell(invokeMsg, getRef());
-
- //verify response msg
- Boolean getMsg = new ExpectMsg<Boolean>("RpcResponse") {
- protected Boolean match(Object in) {
- if (in instanceof RpcResponse) {
- return true;
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
-
- Assert.assertTrue(getMsg);
- }};
- }
-
- @Test
- public void testInvokeRoutedRpcError() throws URISyntaxException {
- new JavaTestKit(system) {{
- ActorRef rpcRegistry = system.actorOf(RpcRegistryOld.props(Mockito.mock(ClusterWrapper.class)));
- Broker.ProviderSession brokerSession = Mockito.mock(Broker.ProviderSession.class);
- SchemaContext schemaContext = mock(SchemaContext.class);
- ActorRef rpcBroker = system.actorOf(RpcBroker.props(brokerSession, rpcRegistry, schemaContext));
- QName rpc = new QName(new URI("actor1"), "actor1");
- CompositeNode input = new ImmutableCompositeNode(QName.create("ns", "2013-12-09", "child1"), new ArrayList<Node<?>>(), ModifyAction.REPLACE);
- InvokeRoutedRpc invokeMsg = new InvokeRoutedRpc(rpc, YangInstanceIdentifier.create(new YangInstanceIdentifier.NodeIdentifier(rpc)), input);
- rpcBroker.tell(invokeMsg, getRef());
+ InvokeRpc invokeMsg = new InvokeRpc(rpc, null, input);
+ rpcBroker1.tell(invokeMsg, getRef());
- Boolean getMsg = new ExpectMsg<Boolean>("ErrorResponse") {
- protected Boolean match(Object in) {
- if (in instanceof ErrorResponse) {
- ErrorResponse reply = (ErrorResponse)in;
- return "No remote actor found for rpc execution.".equals(reply.getException().getMessage());
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
-
- Assert.assertTrue(getMsg);
- }};
- }
+ probeReg1.expectMsgClass(RpcRegistry.Messages.FindRouters.class);
+ List<Pair<ActorRef, Long>> routerList = new ArrayList<Pair<ActorRef, Long>>();
- /**
- * This test method invokes and executes the remote routed rpc
- */
+ routerList.add(new Pair<ActorRef, Long>(rpcBroker2, 200L));
- @Test
- public void testInvokeRoutedRpc() throws URISyntaxException {
- new JavaTestKit(system) {{
- ActorRef rpcRegistry = system.actorOf(RpcRegistryOld.props(mock(ClusterWrapper.class)));
- Broker.ProviderSession brokerSession = mock(Broker.ProviderSession.class);
- SchemaContext schemaContext = mock(SchemaContext.class);
- ActorRef rpcBroker = system.actorOf(RpcBroker.props(brokerSession, rpcRegistry, schemaContext));
- ActorRef rpcBrokerRemote = system.actorOf(RpcBroker.props(brokerSession, rpcRegistry, schemaContext), "actor2");
- // Add Routed RPC in table
- QName rpc = new QName(new URI("actor2"), "actor2");
- YangInstanceIdentifier identifier = YangInstanceIdentifier.create(new YangInstanceIdentifier.NodeIdentifier(rpc));
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, rpc, identifier);
- final String route = rpcBrokerRemote.path().toString();
- Set<RpcRouter.RouteIdentifier<?, ?, ?>> routeIds = new HashSet<>();
- routeIds.add(routeId);
-
- AddRoutedRpc rpcMsg = new AddRoutedRpc(routeIds, route);
- rpcRegistry.tell(rpcMsg, getRef());
- expectMsgEquals(duration("2 second"), "Success");
+ probeReg1.reply(new RpcRegistry.Messages.FindRoutersReply(routerList));
- // invoke rpc
- CompositeNode input = new ImmutableCompositeNode(QName.create("ns", "2013-12-09", "child1"), new ArrayList<Node<?>>(), ModifyAction.REPLACE);
CompositeNode invokeRpcResult = mock(CompositeNode.class);
Collection<RpcError> errors = new ArrayList<>();
RpcResult<CompositeNode> result = Rpcs.getRpcResult(true, invokeRpcResult, errors);
Future<RpcResult<CompositeNode>> rpcResult = Futures.immediateFuture(result);
when(brokerSession.rpc(rpc, input)).thenReturn(rpcResult);
- InvokeRoutedRpc invokeMsg = new InvokeRoutedRpc(rpc, identifier, input);
- rpcBroker.tell(invokeMsg, getRef());
//verify response msg
Boolean getMsg = new ExpectMsg<Boolean>("RpcResponse") {
Assert.assertTrue(getMsg);
}};
}
-
}
--- /dev/null
+/*
+ * Copyright (c) 2014 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 akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.testkit.JavaTestKit;
+import com.typesafe.config.ConfigFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry;
+import org.opendaylight.yangtools.yang.common.QName;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+
+public class RpcListenerTest {
+
+ static ActorSystem system;
+
+
+ @BeforeClass
+ public static void setup() throws InterruptedException {
+ system = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("odl-cluster"));
+ }
+
+ @AfterClass
+ public static void teardown() {
+ JavaTestKit.shutdownActorSystem(system);
+ system = null;
+ }
+
+ @Test
+ public void testRpcAdd() throws URISyntaxException {
+ new JavaTestKit(system) {
+ {
+ JavaTestKit probeReg = new JavaTestKit(system);
+ ActorRef rpcRegistry = probeReg.getRef();
+
+ RpcListener rpcListener = new RpcListener(rpcRegistry);
+
+ QName qName = new QName(new URI("actor2"), "actor2");
+
+ rpcListener.onRpcImplementationAdded(qName);
+ probeReg.expectMsgClass(RpcRegistry.Messages.AddOrUpdateRoutes.class);
+ }};
+
+ }
+
+ @Test
+ public void testRpcRemove() throws URISyntaxException {
+ new JavaTestKit(system) {
+ {
+ JavaTestKit probeReg = new JavaTestKit(system);
+ ActorRef rpcRegistry = probeReg.getRef();
+
+ RpcListener rpcListener = new RpcListener(rpcRegistry);
+
+ QName qName = new QName(new URI("actor2"), "actor2");
+
+ rpcListener.onRpcImplementationRemoved(qName);
+ probeReg.expectMsgClass(RpcRegistry.Messages.RemoveRoutes.class);
+ }};
+
+ }
+}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.registry;
-
-import junit.framework.Assert;
-import org.junit.Test;
-import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
-import org.opendaylight.controller.sal.connector.api.RpcRouter;
-import org.opendaylight.yangtools.yang.common.QName;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.HashSet;
-import java.util.Set;
-
-public class RoutingTableOldTest {
-
- private RoutingTableOld<RpcRouter.RouteIdentifier<?, ?, ?>, String> routingTable =
- new RoutingTableOld<>();
-
- @Test
- public void addGlobalRouteNullRouteIdTest() {
- try {
- routingTable.addGlobalRoute(null, null);
-
- Assert.fail("Null pointer exception was not thrown.");
- } catch (Exception e) {
- Assert.assertEquals(NullPointerException.class.getName(), e.getClass().getName());
- Assert.assertEquals("addGlobalRoute: routeId cannot be null!", e.getMessage());
- }
- }
-
- @Test
- public void addGlobalRouteNullRouteTest() {
- try {
- QName type = new QName(new URI("actor1"), "actor1");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
- routingTable.addGlobalRoute(routeId, null);
-
- Assert.fail("Null pointer exception was not thrown.");
- } catch (Exception e) {
- Assert.assertEquals(NullPointerException.class.getName(), e.getClass().getName());
- Assert.assertEquals("addGlobalRoute: route cannot be null!", e.getMessage());
- }
- }
-
- @Test
- public void getGlobalRouteNullTest() {
- try {
- routingTable.getGlobalRoute(null);
-
- Assert.fail("Null pointer exception was not thrown.");
- } catch (Exception e) {
- Assert.assertEquals(NullPointerException.class.getName(), e.getClass().getName());
- Assert.assertEquals("getGlobalRoute: routeId cannot be null!", e.getMessage());
- }
- }
-
- @Test
- public void getGlobalRouteTest() throws URISyntaxException {
- QName type = new QName(new URI("actor1"), "actor1");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
- String route = "actor1";
-
- routingTable.addGlobalRoute(routeId, route);
-
- String returnedRoute = routingTable.getGlobalRoute(routeId);
-
- Assert.assertEquals(route, returnedRoute);
-
- }
-
- @Test
- public void removeGlobalRouteTest() throws URISyntaxException {
- QName type = new QName(new URI("actorRemove"), "actorRemove");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
- String route = "actorRemove";
-
- routingTable.addGlobalRoute(routeId, route);
-
- String returnedRoute = routingTable.getGlobalRoute(routeId);
-
- Assert.assertEquals(route, returnedRoute);
-
- routingTable.removeGlobalRoute(routeId);
-
- String deletedRoute = routingTable.getGlobalRoute(routeId);
-
- Assert.assertNull(deletedRoute);
- }
-
- @Test
- public void addRoutedRpcNullRouteIdTest() {
- try {
- routingTable.addRoutedRpc(null, null);
-
- Assert.fail("Null pointer exception was not thrown.");
- } catch (Exception e) {
- Assert.assertEquals(NullPointerException.class.getName(), e.getClass().getName());
- Assert.assertEquals("addRoute: routeId cannot be null", e.getMessage());
- }
- }
-
- @Test
- public void addRoutedRpcNullRouteTest() {
- try {
- QName type = new QName(new URI("actor1"), "actor1");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
-
- routingTable.addRoutedRpc(routeId, null);
-
- Assert.fail("Null pointer exception was not thrown.");
- } catch (Exception e) {
- Assert.assertEquals(NullPointerException.class.getName(), e.getClass().getName());
- Assert.assertEquals("addRoute: route cannot be null", e.getMessage());
- }
- }
-
- @Test
- public void getRoutedRpcNullTest() {
- try {
- routingTable.getRoutedRpc(null);
-
- Assert.fail("Null pointer exception was not thrown.");
- } catch (Exception e) {
- Assert.assertEquals(NullPointerException.class.getName(), e.getClass().getName());
- Assert.assertEquals("getRoutes: routeId cannot be null!", e.getMessage());
- }
- }
-
- @Test
- public void getRoutedRpcTest() throws URISyntaxException {
- QName type = new QName(new URI("actor1"), "actor1");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
- String route = "actor1";
-
- routingTable.addRoutedRpc(routeId, route);
-
- Set<String> routes = routingTable.getRoutedRpc(routeId);
-
- Assert.assertEquals(1, routes.size());
- Assert.assertTrue(routes.contains(route));
-
- }
-
- @Test
- public void getLastRoutedRpcTest() throws URISyntaxException {
- QName type = new QName(new URI("first1"), "first1");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
- String route = "first1";
-
- routingTable.addRoutedRpc(routeId, route);
-
- String route2 = "second1";
- routingTable.addRoutedRpc(routeId, route2);
-
- String latest = routingTable.getLastAddedRoutedRpc(routeId);
- Assert.assertEquals(route2, latest);
-
- }
-
- @Test
- public void removeRoutedRpcTest() throws URISyntaxException {
- QName type = new QName(new URI("remove"), "remove");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
- String route = "remove";
- routingTable.addRoutedRpc(routeId, route);
-
- String latest = routingTable.getLastAddedRoutedRpc(routeId);
- Assert.assertEquals(route, latest);
-
- routingTable.removeRoute(routeId, route);
- String removed = routingTable.getLastAddedRoutedRpc(routeId);
- Assert.assertNull(removed);
- }
-
- @Test
- public void removeRoutedRpcsTest() throws URISyntaxException {
- QName type = new QName(new URI("remove1"), "remove1");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
-
- QName type2 = new QName(new URI("remove2"), "remove2");
- RouteIdentifierImpl routeId2 = new RouteIdentifierImpl(null, type2, null);
-
- Set<RpcRouter.RouteIdentifier<?, ?, ?>> routeIds = new HashSet<>();
- routeIds.add(routeId);
- routeIds.add(routeId2);
- String route = "remove1";
-
- routingTable.addRoutedRpcs(routeIds, route);
- String latest1 = routingTable.getLastAddedRoutedRpc(routeId);
- Assert.assertEquals(route, latest1);
-
- String latest2 = routingTable.getLastAddedRoutedRpc(routeId2);
- Assert.assertEquals(route, latest2);
-
- routingTable.removeRoutes(routeIds, route);
- String removed1 = routingTable.getLastAddedRoutedRpc(routeId);
- Assert.assertNull(removed1);
-
- String removed2 = routingTable.getLastAddedRoutedRpc(routeId2);
- Assert.assertNull(removed2);
- }
-
-}
+++ /dev/null
-/*
- * Copyright (c) 2014 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.registry;
-
-import akka.actor.ActorRef;
-import akka.actor.ActorSystem;
-import akka.testkit.JavaTestKit;
-import junit.framework.Assert;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.mockito.Mockito;
-import org.opendaylight.controller.remote.rpc.RouteIdentifierImpl;
-import org.opendaylight.controller.remote.rpc.messages.AddRoutedRpc;
-import org.opendaylight.controller.remote.rpc.messages.AddRpc;
-import org.opendaylight.controller.remote.rpc.messages.GetRoutedRpc;
-import org.opendaylight.controller.remote.rpc.messages.GetRoutedRpcReply;
-import org.opendaylight.controller.remote.rpc.messages.GetRpc;
-import org.opendaylight.controller.remote.rpc.messages.GetRpcReply;
-import org.opendaylight.controller.remote.rpc.messages.RemoveRoutedRpc;
-import org.opendaylight.controller.remote.rpc.messages.RemoveRpc;
-import org.opendaylight.controller.sal.connector.api.RpcRouter;
-import org.opendaylight.yangtools.yang.common.QName;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.HashSet;
-import java.util.Set;
-
-public class RpcRegistryOldTest {
-
- static ActorSystem system;
-
-
- @BeforeClass
- public static void setup() {
- system = ActorSystem.create();
- }
-
- @AfterClass
- public static void teardown() {
- JavaTestKit.shutdownActorSystem(system);
- system = null;
- }
-
- /**
- This test add, read and remove an entry in global rpc
- */
- @Test
- public void testGlobalRpc() throws URISyntaxException {
- new JavaTestKit(system) {{
- ActorRef rpcRegistry = system.actorOf(RpcRegistryOld.props(Mockito.mock(ClusterWrapper.class)));
- QName type = new QName(new URI("actor1"), "actor1");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
- final String route = "actor1";
-
- AddRpc rpcMsg = new AddRpc(routeId, route);
- rpcRegistry.tell(rpcMsg, getRef());
- expectMsgEquals(duration("2 second"), "Success");
-
- GetRpc getRpc = new GetRpc(routeId);
- rpcRegistry.tell(getRpc, getRef());
-
- Boolean getMsg = new ExpectMsg<Boolean>("GetRpcReply") {
- protected Boolean match(Object in) {
- if (in instanceof GetRpcReply) {
- GetRpcReply reply = (GetRpcReply)in;
- return route.equals(reply.getRoutePath());
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
-
- Assert.assertTrue(getMsg);
-
- RemoveRpc removeMsg = new RemoveRpc(routeId);
- rpcRegistry.tell(removeMsg, getRef());
- expectMsgEquals(duration("2 second"), "Success");
-
- rpcRegistry.tell(getRpc, getRef());
-
- Boolean getNullMsg = new ExpectMsg<Boolean>("GetRpcReply") {
- protected Boolean match(Object in) {
- if (in instanceof GetRpcReply) {
- GetRpcReply reply = (GetRpcReply)in;
- return reply.getRoutePath() == null;
- } else {
- throw noMatch();
- }
- }
- }.get();
- Assert.assertTrue(getNullMsg);
- }};
-
- }
-
- /**
- This test add, read and remove an entry in routed rpc
- */
- @Test
- public void testRoutedRpc() throws URISyntaxException {
- new JavaTestKit(system) {{
- ActorRef rpcRegistry = system.actorOf(RpcRegistryOld.props(Mockito.mock(ClusterWrapper.class)));
- QName type = new QName(new URI("actor1"), "actor1");
- RouteIdentifierImpl routeId = new RouteIdentifierImpl(null, type, null);
- final String route = "actor1";
-
- Set<RpcRouter.RouteIdentifier<?, ?, ?>> routeIds = new HashSet<>();
- routeIds.add(routeId);
-
- AddRoutedRpc rpcMsg = new AddRoutedRpc(routeIds, route);
- rpcRegistry.tell(rpcMsg, getRef());
- expectMsgEquals(duration("2 second"), "Success");
-
- GetRoutedRpc getRpc = new GetRoutedRpc(routeId);
- rpcRegistry.tell(getRpc, getRef());
-
- Boolean getMsg = new ExpectMsg<Boolean>("GetRoutedRpcReply") {
- protected Boolean match(Object in) {
- if (in instanceof GetRoutedRpcReply) {
- GetRoutedRpcReply reply = (GetRoutedRpcReply)in;
- return route.equals(reply.getRoutePath());
- } else {
- throw noMatch();
- }
- }
- }.get(); // this extracts the received message
-
- Assert.assertTrue(getMsg);
-
- RemoveRoutedRpc removeMsg = new RemoveRoutedRpc(routeIds, route);
- rpcRegistry.tell(removeMsg, getRef());
- expectMsgEquals(duration("2 second"), "Success");
-
- rpcRegistry.tell(getRpc, getRef());
-
- Boolean getNullMsg = new ExpectMsg<Boolean>("GetRoutedRpcReply") {
- protected Boolean match(Object in) {
- if (in instanceof GetRoutedRpcReply) {
- GetRoutedRpcReply reply = (GetRoutedRpcReply)in;
- return reply.getRoutePath() == null;
- } else {
- throw noMatch();
- }
- }
- }.get();
- Assert.assertTrue(getNullMsg);
- }};
-
- }
-
-}
@BeforeClass
public static void setup() {
- system = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("unit-test"));
+ system = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("odl-cluster"));
system.actorOf(Props.create(TerminationMonitor.class), "termination-monitor");
store = createStore();
@BeforeClass
public static void setup() throws InterruptedException {
- Thread.sleep(1000);//give some time for previous test to stop the system. Netty port conflict arises otherwise.
- system = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("unit-test"));
+ system = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("odl-cluster"));
system.actorOf(Props.create(TerminationMonitor.class), "termination-monitor");
gossiper = createGossiper();
--- /dev/null
+/*
+ * Copyright (c) 2014 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.utils;
+
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.japi.Pair;
+import akka.testkit.JavaTestKit;
+import akka.testkit.TestProbe;
+import com.typesafe.config.ConfigFactory;
+import junit.framework.Assert;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+
+public class LatestEntryRoutingLogicTest {
+
+ static ActorSystem system;
+
+ @BeforeClass
+ public static void setup() throws InterruptedException {
+ system = ActorSystem.create("opendaylight-rpc", ConfigFactory.load().getConfig("odl-cluster"));
+ }
+
+ @AfterClass
+ public static void teardown() {
+ JavaTestKit.shutdownActorSystem(system);
+ system = null;
+ }
+
+ @Test
+ public void testRoutingLogic() {
+ List<Pair<ActorRef, Long>> pairList = new ArrayList<>();
+ TestProbe probe1 = new TestProbe(system);
+ TestProbe probe2 = new TestProbe(system);
+ TestProbe probe3 = new TestProbe(system);
+ ActorRef actor1 = probe1.ref();
+ ActorRef actor2 = probe2.ref();
+ ActorRef actor3 = probe3.ref();
+ pairList.add(new Pair<ActorRef, Long>(actor1, 1000L));
+ pairList.add(new Pair<ActorRef, Long>(actor2, 3000L));
+ pairList.add(new Pair<ActorRef, Long>(actor3, 2000L));
+ RoutingLogic logic = new LatestEntryRoutingLogic(pairList);
+ Assert.assertTrue(logic.select().equals(actor2));
+ }
+}
log-remote-lifecycle-events = off
netty.tcp {
hostname = "127.0.0.1"
- port = 2551
+ port = 2550
}
}
cluster {
- seed-nodes = ["akka.tcp://opendaylight-rpc@127.0.0.1:2551"]
+ seed-nodes = ["akka.tcp://opendaylight-rpc@127.0.0.1:2550"]
auto-down-unreachable-after = 10s
}
Code Review / controller.git / commitdiff