X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-binding-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fsal%2Fbinding%2Fimpl%2FRpcProviderRegistryImpl.java;h=0949d3d7612dfb34a1f8890329653af7efebc35e;hp=e98d5b9942c86afb9b76e006662fe02cdf6147c2;hb=3591817114661bb7971d6d355186ff1b39636fcd;hpb=e962193b5d42003b8e2b94bd205c9fd54f3f41bb diff --git a/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/RpcProviderRegistryImpl.java b/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/RpcProviderRegistryImpl.java index e98d5b9942..0949d3d761 100644 --- a/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/RpcProviderRegistryImpl.java +++ b/opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/RpcProviderRegistryImpl.java @@ -8,14 +8,21 @@ package org.opendaylight.controller.sal.binding.impl; import static com.google.common.base.Preconditions.checkState; - +import com.google.common.base.Preconditions; +import com.google.common.base.Throwables; +import com.google.common.cache.Cache; +import com.google.common.cache.CacheBuilder; +import com.google.common.cache.CacheLoader; +import com.google.common.cache.LoadingCache; +import com.google.common.util.concurrent.UncheckedExecutionException; import java.util.EventListener; import java.util.HashMap; import java.util.Map; import java.util.Map.Entry; import java.util.Set; -import java.util.WeakHashMap; - +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicBoolean; import org.opendaylight.controller.md.sal.common.api.routing.RouteChange; import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener; import org.opendaylight.controller.md.sal.common.api.routing.RouteChangePublisher; @@ -25,26 +32,37 @@ import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistr import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry; import org.opendaylight.controller.sal.binding.api.rpc.RpcContextIdentifier; import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter; +import org.opendaylight.controller.sal.binding.codegen.RpcIsNotRoutedException; import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator; import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeHelper; import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder; import org.opendaylight.yangtools.concepts.AbstractObjectRegistration; import org.opendaylight.yangtools.concepts.ListenerRegistration; -import org.opendaylight.yangtools.concepts.util.ListenerRegistry; +import org.opendaylight.yangtools.util.ListenerRegistry; import org.opendaylight.yangtools.yang.binding.BaseIdentity; import org.opendaylight.yangtools.yang.binding.InstanceIdentifier; import org.opendaylight.yangtools.yang.binding.RpcService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class RpcProviderRegistryImpl implements // - RpcProviderRegistry, // - RouteChangePublisher> { +public class RpcProviderRegistryImpl implements RpcProviderRegistry, RouteChangePublisher> { private RuntimeCodeGenerator rpcFactory = SingletonHolder.RPC_GENERATOR_IMPL; - private final Map, RpcService> publicProxies = new WeakHashMap<>(); - private final Map, RpcRouter> rpcRouters = new WeakHashMap<>(); + // cache of proxy objects where each value in the map corresponds to a specific RpcService + private final LoadingCache, RpcService> publicProxies = CacheBuilder.newBuilder().weakKeys(). + build(new CacheLoader, RpcService>() { + @Override + public RpcService load(final Class type) { + final RpcService proxy = rpcFactory.getDirectProxyFor(type); + LOG.debug("Created {} as public proxy for {} in {}", proxy, type.getSimpleName(), this); + return proxy; + } + }); + + private final Cache, RpcRouter> rpcRouters = CacheBuilder.newBuilder().weakKeys() + .build(); + private final ListenerRegistry>> routeChangeListeners = ListenerRegistry .create(); private final ListenerRegistry routerInstantiationListener = ListenerRegistry.create(); @@ -59,26 +77,33 @@ public class RpcProviderRegistryImpl implements // return name; } - public RpcProviderRegistryImpl(String name) { + public RpcProviderRegistryImpl(final String name) { super(); this.name = name; } @Override - public final RoutedRpcRegistration addRoutedRpcImplementation(Class type, - T implementation) throws IllegalStateException { + public final RoutedRpcRegistration addRoutedRpcImplementation(final Class type, + final T implementation) throws IllegalStateException { return getRpcRouter(type).addRoutedRpcImplementation(implementation); } @Override - public final RpcRegistration addRpcImplementation(Class type, T implementation) - throws IllegalStateException { - @SuppressWarnings("unchecked") - RpcRouter potentialRouter = (RpcRouter) rpcRouters.get(type); - if (potentialRouter != null) { + public final RpcRegistration addRpcImplementation(final Class type, final T implementation) { + + // FIXME: This should be well documented - addRpcImplementation for + // routed RPCs + try { + // Note: If RPC is really global, expected count of registrations + // of this method is really low. + RpcRouter potentialRouter = getRpcRouter(type); checkState(potentialRouter.getDefaultService() == null, - "Default service for routed RPC already registered."); + "Default service for routed RPC already registered."); return potentialRouter.registerDefaultService(implementation); + } catch (RpcIsNotRoutedException e) { + // NOOP - we could safely continue, since RPC is not routed + // so we fallback to global routing. + LOG.debug("RPC is not routed. Using global registration.",e); } T publicProxy = getRpcService(type); RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy); @@ -91,55 +116,45 @@ public class RpcProviderRegistryImpl implements // @SuppressWarnings("unchecked") @Override - public final T getRpcService(Class type) { - - T potentialProxy = (T) publicProxies.get(type); - if (potentialProxy != null) { - return potentialProxy; - } - synchronized (this) { - /** - * Potential proxy could be instantiated by other thread while we - * were waiting for the lock. - */ - - potentialProxy = (T) publicProxies.get(type); - if (potentialProxy != null) { - return potentialProxy; - } - T proxy = rpcFactory.getDirectProxyFor(type); - LOG.debug("Created {} as public proxy for {} in {}", proxy, type.getSimpleName(), this); - publicProxies.put(type, proxy); - return proxy; - } + public final T getRpcService(final Class type) { + return (T) publicProxies.getUnchecked(type); } - @SuppressWarnings("unchecked") - public RpcRouter getRpcRouter(Class type) { - RpcRouter potentialRouter = rpcRouters.get(type); - if (potentialRouter != null) { - return (RpcRouter) potentialRouter; - } - synchronized (this) { - /** - * Potential Router could be instantiated by other thread while we - * were waiting for the lock. - */ - potentialRouter = rpcRouters.get(type); - if (potentialRouter != null) { - return (RpcRouter) potentialRouter; + + public RpcRouter getRpcRouter(final Class type) { + try { + final AtomicBoolean created = new AtomicBoolean(false); + @SuppressWarnings( "unchecked") + // LoadingCache is unsuitable for RpcRouter since we need to distinguish + // first creation of RPC Router, so that is why + // we are using normal cache with load API and shared AtomicBoolean + // for this call, which will be set to true if router was created. + RpcRouter router = (RpcRouter) rpcRouters.get(type,new Callable>() { + + @Override + public org.opendaylight.controller.sal.binding.api.rpc.RpcRouter call() { + RpcRouter router = rpcFactory.getRouterFor(type, name); + router.registerRouteChangeListener(new RouteChangeForwarder(type)); + LOG.debug("Registering router {} as global implementation of {} in {}", router, type.getSimpleName(), this); + RuntimeCodeHelper.setDelegate(getRpcService(type), router.getInvocationProxy()); + created.set(true); + return router; + } + }); + if(created.get()) { + notifyListenersRoutedCreated(router); } - RpcRouter router = rpcFactory.getRouterFor(type, name); - router.registerRouteChangeListener(new RouteChangeForwarder(type)); - LOG.debug("Registering router {} as global implementation of {} in {}", router, type.getSimpleName(), this); - RuntimeCodeHelper.setDelegate(getRpcService(type), router.getInvocationProxy()); - rpcRouters.put(type, router); - notifyListenersRoutedCreated(router); return router; + } catch (ExecutionException | UncheckedExecutionException e) { + // We rethrow Runtime Exceptions which were wrapped by + // Execution Exceptions + // otherwise we throw IllegalStateException with original + Throwables.propagateIfPossible(e.getCause()); + throw new IllegalStateException("Could not load RPC Router for "+type.getName(),e); } } - private void notifyGlobalRpcAdded(Class type) { + private void notifyGlobalRpcAdded(final Class type) { for(ListenerRegistration listener : globalRpcListeners) { try { listener.getInstance().onGlobalRpcRegistered(type); @@ -150,7 +165,7 @@ public class RpcProviderRegistryImpl implements // } - private void notifyListenersRoutedCreated(RpcRouter router) { + private void notifyListenersRoutedCreated(final RpcRouter router) { for (ListenerRegistration listener : routerInstantiationListener) { try { @@ -163,10 +178,10 @@ public class RpcProviderRegistryImpl implements // } public ListenerRegistration registerRouterInstantiationListener( - RouterInstantiationListener listener) { + final RouterInstantiationListener listener) { ListenerRegistration reg = routerInstantiationListener.register(listener); try { - for (RpcRouter router : rpcRouters.values()) { + for (RpcRouter router : rpcRouters.asMap().values()) { listener.onRpcRouterCreated(router); } } catch (Exception e) { @@ -175,9 +190,10 @@ public class RpcProviderRegistryImpl implements // return reg; } + @SuppressWarnings("unchecked") @Override public >> ListenerRegistration registerRouteChangeListener( - L listener) { + final L listener) { return (ListenerRegistration) routeChangeListeners.register(listener); } @@ -185,7 +201,7 @@ public class RpcProviderRegistryImpl implements // return rpcFactory; } - public void setRpcFactory(RuntimeCodeGenerator rpcFactory) { + public void setRpcFactory(final RuntimeCodeGenerator rpcFactory) { this.rpcFactory = rpcFactory; } @@ -193,7 +209,7 @@ public class RpcProviderRegistryImpl implements // void onRpcRouterCreated(RpcRouter router); } - public ListenerRegistration registerGlobalRpcRegistrationListener(GlobalRpcRegistrationListener listener) { + public ListenerRegistration registerGlobalRpcRegistrationListener(final GlobalRpcRegistrationListener listener) { return globalRpcListeners.register(listener); } @@ -203,17 +219,15 @@ public class RpcProviderRegistryImpl implements // } - private class RouteChangeForwarder implements - RouteChangeListener, InstanceIdentifier> { - + private final class RouteChangeForwarder implements RouteChangeListener, InstanceIdentifier> { private final Class type; - public RouteChangeForwarder(Class type) { + RouteChangeForwarder(final Class type) { this.type = type; } @Override - public void onRouteChange(RouteChange, InstanceIdentifier> change) { + public void onRouteChange(final RouteChange, InstanceIdentifier> change) { Map>> announcements = new HashMap<>(); for (Entry, Set>> entry : change.getAnnouncements() .entrySet()) { @@ -232,21 +246,20 @@ public class RpcProviderRegistryImpl implements // try { listener.getInstance().onRouteChange(toPublish); } catch (Exception e) { - e.printStackTrace(); + LOG.error("Unhandled exception during invoking listener",listener.getInstance(),e); } } } } - public static class RpcProxyRegistration extends AbstractObjectRegistration implements - RpcRegistration { - + private static final class RpcProxyRegistration extends AbstractObjectRegistration implements RpcRegistration { + private final RpcProviderRegistryImpl registry; private final Class serviceType; - private RpcProviderRegistryImpl registry; - public RpcProxyRegistration(Class type, T service, RpcProviderRegistryImpl registry) { + RpcProxyRegistration(final Class type, final T service, final RpcProviderRegistryImpl registry) { super(service); - serviceType = type; + this.registry = Preconditions.checkNotNull(registry); + this.serviceType = type; } @Override @@ -256,13 +269,10 @@ public class RpcProviderRegistryImpl implements // @Override protected void removeRegistration() { - if (registry != null) { - T publicProxy = registry.getRpcService(serviceType); - RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy); - if (currentDelegate == getInstance()) { - RuntimeCodeHelper.setDelegate(publicProxy, null); - } - registry = null; + T publicProxy = registry.getRpcService(serviceType); + RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy); + if (currentDelegate == getInstance()) { + RuntimeCodeHelper.setDelegate(publicProxy, null); } } }