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%2Fregistry%2FRoutingTable.java;h=62312c14cb90cc60fed0cdec0435a050f8ad16ef;hp=5e19653a22d21ed83ae3eec370290dc37f50c1ce;hb=HEAD;hpb=17d82f582a6bc13c78be3b19954ff8c021180e93 diff --git a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/RoutingTable.java b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/RoutingTable.java index 5e19653a22..f0795a3a42 100644 --- a/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/RoutingTable.java +++ b/opendaylight/md-sal/sal-remoterpc-connector/src/main/java/org/opendaylight/controller/remote/rpc/registry/RoutingTable.java @@ -5,167 +5,99 @@ * 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 akka.actor.ActorRef; +import akka.serialization.JavaSerializer; +import akka.serialization.Serialization; +import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashSet; import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; - -public class RoutingTable { - - private final Logger LOG = LoggerFactory.getLogger(RoutingTable.class); - - private ConcurrentMap globalRpcMap = new ConcurrentHashMap<>(); - private ConcurrentMap> routedRpcMap = new ConcurrentHashMap<>(); - - public ConcurrentMap getGlobalRpcMap() { - return globalRpcMap; - } - - public ConcurrentMap> 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); +import org.opendaylight.mdsal.dom.api.DOMRpcIdentifier; +import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeDataInput; +import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeDataOutput; +import org.opendaylight.yangtools.yang.data.codec.binfmt.NormalizedNodeStreamVersion; +import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute; + +public final class RoutingTable extends AbstractRoutingTable { + private static final class Proxy implements Externalizable { + private static final long serialVersionUID = 1L; + + @SuppressFBWarnings(value = "SE_BAD_FIELD", justification = "We deal with the field in serialization methods.") + private Collection rpcs; + private ActorRef opsInvoker; + + // checkstyle flags the public modifier as redundant however it is explicitly needed for Java serialization to + // be able to create instances via reflection. + @SuppressWarnings("checkstyle:RedundantModifier") + public Proxy() { + // For Externalizable + } + + Proxy(final RoutingTable table) { + rpcs = table.getItems(); + opsInvoker = table.getInvoker(); + } + + @Override + public void writeExternal(final ObjectOutput out) throws IOException { + out.writeObject(Serialization.serializedActorPath(opsInvoker)); + + try (NormalizedNodeDataOutput nnout = NormalizedNodeStreamVersion.current().newDataOutput(out)) { + nnout.writeInt(rpcs.size()); + for (DOMRpcIdentifier id : rpcs) { + // TODO: we should be able to get by with just a QName + nnout.writeSchemaNodeIdentifier(Absolute.of(id.getType())); + nnout.writeYangInstanceIdentifier(id.getContextReference()); + } + } + } + + @Override + public void readExternal(final ObjectInput in) throws IOException, ClassNotFoundException { + opsInvoker = JavaSerializer.currentSystem().value().provider().resolveActorRef((String) in.readObject()); + + final NormalizedNodeDataInput nnin = NormalizedNodeDataInput.newDataInput(in); + final int size = nnin.readInt(); + rpcs = new ArrayList<>(size); + for (int i = 0; i < size; ++i) { + // TODO: we should be able to get by with just a QName + rpcs.add(DOMRpcIdentifier.create(nnin.readSchemaNodeIdentifier().firstNodeIdentifier(), + nnin.readYangInstanceIdentifier())); + } + } + + private Object readResolve() { + return new RoutingTable(opsInvoker, rpcs); + } } - } - - 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 getRoutedRpc(final I routeId) { - Preconditions.checkNotNull(routeId, "getRoutes: routeId cannot be null!"); - Set routes = routedRpcMap.get(routeId); + private static final long serialVersionUID = 1L; - if (routes == null) { - return Collections.emptySet(); + RoutingTable(final ActorRef invoker, final Collection table) { + super(invoker, table); } - return ImmutableSet.copyOf(routes); - } - - public R getLastAddedRoutedRpc(final I routeId) { - - Set routes = getRoutedRpc(routeId); - - if (routes.isEmpty()) { - return null; + RoutingTable addRpcs(final Collection toAdd) { + final Set newRpcs = new HashSet<>(getItems()); + newRpcs.addAll(toAdd); + return new RoutingTable(getInvoker(), newRpcs); } - R route = null; - Iterator iter = routes.iterator(); - while (iter.hasNext()) { - route = iter.next(); + RoutingTable removeRpcs(final Collection toRemove) { + final Set newRpcs = new HashSet<>(getItems()); + newRpcs.removeAll(toRemove); + return new RoutingTable(getInvoker(), newRpcs); } - 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 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 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 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 updatedRoutes = new LinkedHashSet<>(); - updatedRoutes.add(route); - LinkedHashSet 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 updatedRoutes = null; - for (int i=0;i<100;i++){ - LinkedHashSet 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; - } - + @Override + Object writeReplace() { + return new Proxy(this); } - //the method did not already return means it failed to remove route in 100 attempts - throw new IllegalStateException("Failed to remove route [" + routeId + "]"); - } }