2 * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.controller.sal.binding.impl;
10 import static com.google.common.base.Preconditions.checkState;
11 import com.google.common.base.Preconditions;
12 import com.google.common.base.Throwables;
13 import com.google.common.cache.Cache;
14 import com.google.common.cache.CacheBuilder;
15 import com.google.common.cache.CacheLoader;
16 import com.google.common.cache.LoadingCache;
17 import com.google.common.util.concurrent.UncheckedExecutionException;
18 import java.util.EventListener;
19 import java.util.HashMap;
21 import java.util.Map.Entry;
23 import java.util.concurrent.Callable;
24 import java.util.concurrent.ExecutionException;
25 import java.util.concurrent.atomic.AtomicBoolean;
26 import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
27 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
28 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangePublisher;
29 import org.opendaylight.controller.md.sal.common.impl.routing.RoutingUtils;
30 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RoutedRpcRegistration;
31 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
32 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
33 import org.opendaylight.controller.sal.binding.api.rpc.RpcContextIdentifier;
34 import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter;
35 import org.opendaylight.controller.sal.binding.codegen.RpcIsNotRoutedException;
36 import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator;
37 import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeHelper;
38 import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
39 import org.opendaylight.yangtools.concepts.AbstractObjectRegistration;
40 import org.opendaylight.yangtools.concepts.ListenerRegistration;
41 import org.opendaylight.yangtools.util.ListenerRegistry;
42 import org.opendaylight.yangtools.yang.binding.BaseIdentity;
43 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
44 import org.opendaylight.yangtools.yang.binding.RpcService;
45 import org.slf4j.Logger;
46 import org.slf4j.LoggerFactory;
48 public class RpcProviderRegistryImpl implements RpcProviderRegistry, RouteChangePublisher<RpcContextIdentifier, InstanceIdentifier<?>> {
50 private RuntimeCodeGenerator rpcFactory = SingletonHolder.RPC_GENERATOR_IMPL;
52 // cache of proxy objects where each value in the map corresponds to a specific RpcService
53 private final LoadingCache<Class<? extends RpcService>, RpcService> publicProxies = CacheBuilder.newBuilder().weakKeys().
54 build(new CacheLoader<Class<? extends RpcService>, RpcService>() {
56 public RpcService load(final Class<? extends RpcService> type) {
57 final RpcService proxy = rpcFactory.getDirectProxyFor(type);
58 LOG.debug("Created {} as public proxy for {} in {}", proxy, type.getSimpleName(), this);
63 private final Cache<Class<? extends RpcService>, RpcRouter<?>> rpcRouters = CacheBuilder.newBuilder().weakKeys()
66 private final ListenerRegistry<RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> routeChangeListeners = ListenerRegistry
68 private final ListenerRegistry<RouterInstantiationListener> routerInstantiationListener = ListenerRegistry.create();
70 private final static Logger LOG = LoggerFactory.getLogger(RpcProviderRegistryImpl.class);
72 private final String name;
74 private final ListenerRegistry<GlobalRpcRegistrationListener> globalRpcListeners = ListenerRegistry.create();
76 public String getName() {
80 public RpcProviderRegistryImpl(final String name) {
86 public final <T extends RpcService> RoutedRpcRegistration<T> addRoutedRpcImplementation(final Class<T> type,
87 final T implementation) throws IllegalStateException {
88 return getRpcRouter(type).addRoutedRpcImplementation(implementation);
92 public final <T extends RpcService> RpcRegistration<T> addRpcImplementation(final Class<T> type, final T implementation)
93 throws IllegalStateException {
95 // FIXME: This should be well documented - addRpcImplementation for
98 // Note: If RPC is really global, expected count of registrations
99 // of this method is really low.
100 RpcRouter<T> potentialRouter = getRpcRouter(type);
101 checkState(potentialRouter.getDefaultService() == null,
102 "Default service for routed RPC already registered.");
103 return potentialRouter.registerDefaultService(implementation);
104 } catch (RpcIsNotRoutedException e) {
105 // NOOP - we could safely continue, since RPC is not routed
106 // so we fallback to global routing.
107 LOG.debug("RPC is not routed. Using global registration.",e);
109 T publicProxy = getRpcService(type);
110 RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy);
111 checkState(currentDelegate == null, "Rpc service is already registered");
112 LOG.debug("Registering {} as global implementation of {} in {}", implementation, type.getSimpleName(), this);
113 RuntimeCodeHelper.setDelegate(publicProxy, implementation);
114 notifyGlobalRpcAdded(type);
115 return new RpcProxyRegistration<T>(type, implementation, this);
118 @SuppressWarnings("unchecked")
120 public final <T extends RpcService> T getRpcService(final Class<T> type) {
121 return (T) publicProxies.getUnchecked(type);
125 public <T extends RpcService> RpcRouter<T> getRpcRouter(final Class<T> type) {
127 final AtomicBoolean created = new AtomicBoolean(false);
128 @SuppressWarnings( "unchecked")
129 // LoadingCache is unsuitable for RpcRouter since we need to distinguish
130 // first creation of RPC Router, so that is why
131 // we are using normal cache with load API and shared AtomicBoolean
132 // for this call, which will be set to true if router was created.
133 RpcRouter<T> router = (RpcRouter<T>) rpcRouters.get(type,new Callable<RpcRouter<?>>() {
136 public org.opendaylight.controller.sal.binding.api.rpc.RpcRouter<?> call() {
137 RpcRouter<?> router = rpcFactory.getRouterFor(type, name);
138 router.registerRouteChangeListener(new RouteChangeForwarder<T>(type));
139 LOG.debug("Registering router {} as global implementation of {} in {}", router, type.getSimpleName(), this);
140 RuntimeCodeHelper.setDelegate(getRpcService(type), router.getInvocationProxy());
146 notifyListenersRoutedCreated(router);
149 } catch (ExecutionException | UncheckedExecutionException e) {
150 // We rethrow Runtime Exceptions which were wrapped by
151 // Execution Exceptions
152 // otherwise we throw IllegalStateException with original
153 Throwables.propagateIfPossible(e.getCause());
154 throw new IllegalStateException("Could not load RPC Router for "+type.getName(),e);
158 private void notifyGlobalRpcAdded(final Class<? extends RpcService> type) {
159 for(ListenerRegistration<GlobalRpcRegistrationListener> listener : globalRpcListeners) {
161 listener.getInstance().onGlobalRpcRegistered(type);
162 } catch (Exception e) {
163 LOG.error("Unhandled exception during invoking listener {}", e);
169 private void notifyListenersRoutedCreated(final RpcRouter<?> router) {
171 for (ListenerRegistration<RouterInstantiationListener> listener : routerInstantiationListener) {
173 listener.getInstance().onRpcRouterCreated(router);
174 } catch (Exception e) {
175 LOG.error("Unhandled exception during invoking listener {}", e);
181 public ListenerRegistration<RouterInstantiationListener> registerRouterInstantiationListener(
182 final RouterInstantiationListener listener) {
183 ListenerRegistration<RouterInstantiationListener> reg = routerInstantiationListener.register(listener);
185 for (RpcRouter<?> router : rpcRouters.asMap().values()) {
186 listener.onRpcRouterCreated(router);
188 } catch (Exception e) {
189 LOG.error("Unhandled exception during invoking listener {}", e);
194 @SuppressWarnings("unchecked")
196 public <L extends RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> ListenerRegistration<L> registerRouteChangeListener(
198 return (ListenerRegistration<L>) routeChangeListeners.register(listener);
201 public RuntimeCodeGenerator getRpcFactory() {
205 public void setRpcFactory(final RuntimeCodeGenerator rpcFactory) {
206 this.rpcFactory = rpcFactory;
209 public interface RouterInstantiationListener extends EventListener {
210 void onRpcRouterCreated(RpcRouter<?> router);
213 public ListenerRegistration<GlobalRpcRegistrationListener> registerGlobalRpcRegistrationListener(final GlobalRpcRegistrationListener listener) {
214 return globalRpcListeners.register(listener);
217 public interface GlobalRpcRegistrationListener extends EventListener {
218 void onGlobalRpcRegistered(Class<? extends RpcService> cls);
219 void onGlobalRpcUnregistered(Class<? extends RpcService> cls);
223 private final class RouteChangeForwarder<T extends RpcService> implements RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>> {
224 private final Class<T> type;
226 RouteChangeForwarder(final Class<T> type) {
231 public void onRouteChange(final RouteChange<Class<? extends BaseIdentity>, InstanceIdentifier<?>> change) {
232 Map<RpcContextIdentifier, Set<InstanceIdentifier<?>>> announcements = new HashMap<>();
233 for (Entry<Class<? extends BaseIdentity>, Set<InstanceIdentifier<?>>> entry : change.getAnnouncements()
235 RpcContextIdentifier key = RpcContextIdentifier.contextFor(type, entry.getKey());
236 announcements.put(key, entry.getValue());
238 Map<RpcContextIdentifier, Set<InstanceIdentifier<?>>> removals = new HashMap<>();
239 for (Entry<Class<? extends BaseIdentity>, Set<InstanceIdentifier<?>>> entry : change.getRemovals()
241 RpcContextIdentifier key = RpcContextIdentifier.contextFor(type, entry.getKey());
242 removals.put(key, entry.getValue());
244 RouteChange<RpcContextIdentifier, InstanceIdentifier<?>> toPublish = RoutingUtils
245 .<RpcContextIdentifier, InstanceIdentifier<?>> change(announcements, removals);
246 for (ListenerRegistration<RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> listener : routeChangeListeners) {
248 listener.getInstance().onRouteChange(toPublish);
249 } catch (Exception e) {
250 LOG.error("Unhandled exception during invoking listener",listener.getInstance(),e);
256 private static final class RpcProxyRegistration<T extends RpcService> extends AbstractObjectRegistration<T> implements RpcRegistration<T> {
257 private final RpcProviderRegistryImpl registry;
258 private final Class<T> serviceType;
260 RpcProxyRegistration(final Class<T> type, final T service, final RpcProviderRegistryImpl registry) {
262 this.registry = Preconditions.checkNotNull(registry);
263 this.serviceType = type;
267 public Class<T> getServiceType() {
272 protected void removeRegistration() {
273 T publicProxy = registry.getRpcService(serviceType);
274 RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy);
275 if (currentDelegate == getInstance()) {
276 RuntimeCodeHelper.setDelegate(publicProxy, null);