opendaylight/md-sal/sal-binding-broker/src/main/java/org/opendaylight/controller/sal/binding/impl/RpcProviderRegistryImpl.java

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   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
   7  */
   8 package org.opendaylight.controller.sal.binding.impl;
   9
  10 import static com.google.common.base.Preconditions.checkState;
  11
  12 import java.util.EventListener;
  13 import java.util.HashMap;
  14 import java.util.Map;
  15 import java.util.Map.Entry;
  16 import java.util.Set;
  17 import java.util.concurrent.Callable;
  18 import java.util.concurrent.ExecutionException;
  19 import java.util.concurrent.atomic.AtomicBoolean;
  20
  21 import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
  22 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
  23 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangePublisher;
  24 import org.opendaylight.controller.md.sal.common.impl.routing.RoutingUtils;
  25 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RoutedRpcRegistration;
  26 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
  27 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  28 import org.opendaylight.controller.sal.binding.api.rpc.RpcContextIdentifier;
  29 import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter;
  30 import org.opendaylight.controller.sal.binding.codegen.RpcIsNotRoutedException;
  31 import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator;
  32 import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeHelper;
  33 import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
  34 import org.opendaylight.yangtools.concepts.AbstractObjectRegistration;
  35 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  36 import org.opendaylight.yangtools.concepts.util.ListenerRegistry;
  37 import org.opendaylight.yangtools.yang.binding.BaseIdentity;
  38 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  39 import org.opendaylight.yangtools.yang.binding.RpcService;
  40 import org.slf4j.Logger;
  41 import org.slf4j.LoggerFactory;
  42
  43 import com.google.common.base.Throwables;
  44 import com.google.common.cache.Cache;
  45 import com.google.common.cache.CacheBuilder;
  46 import com.google.common.cache.CacheLoader;
  47 import com.google.common.cache.LoadingCache;
  48 import com.google.common.util.concurrent.UncheckedExecutionException;
  49
  50 public class RpcProviderRegistryImpl implements RpcProviderRegistry, RouteChangePublisher<RpcContextIdentifier, InstanceIdentifier<?>> {
  51
  52     private RuntimeCodeGenerator rpcFactory = SingletonHolder.RPC_GENERATOR_IMPL;
  53
  54     // cache of proxy objects where each value in the map corresponds to a specific RpcService
  55     private final LoadingCache<Class<? extends RpcService>, RpcService> publicProxies = CacheBuilder.newBuilder().weakKeys().
  56             build(new CacheLoader<Class<? extends RpcService>, RpcService>() {
  57                 @Override
  58                 public RpcService load(final Class<? extends RpcService> type) {
  59                     final RpcService proxy = rpcFactory.getDirectProxyFor(type);
  60                     LOG.debug("Created {} as public proxy for {} in {}", proxy, type.getSimpleName(), this);
  61                     return proxy;
  62                 }
  63             });
  64
  65     private final Cache<Class<? extends RpcService>, RpcRouter<?>> rpcRouters = CacheBuilder.newBuilder().weakKeys()
  66             .build();
  67
  68     private final ListenerRegistry<RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> routeChangeListeners = ListenerRegistry
  69             .create();
  70     private final ListenerRegistry<RouterInstantiationListener> routerInstantiationListener = ListenerRegistry.create();
  71
  72     private final static Logger LOG = LoggerFactory.getLogger(RpcProviderRegistryImpl.class);
  73
  74     private final String name;
  75
  76     private final ListenerRegistry<GlobalRpcRegistrationListener> globalRpcListeners = ListenerRegistry.create();
  77
  78     public String getName() {
  79         return name;
  80     }
  81
  82     public RpcProviderRegistryImpl(final String name) {
  83         super();
  84         this.name = name;
  85     }
  86
  87     @Override
  88     public final <T extends RpcService> RoutedRpcRegistration<T> addRoutedRpcImplementation(final Class<T> type,
  89             final T implementation) throws IllegalStateException {
  90         return getRpcRouter(type).addRoutedRpcImplementation(implementation);
  91     }
  92
  93     @Override
  94     public final <T extends RpcService> RpcRegistration<T> addRpcImplementation(final Class<T> type, final T implementation)
  95             throws IllegalStateException {
  96
  97         // FIXME: This should be well documented - addRpcImplementation for
  98         // routed RPCs
  99         try {
 100             // Note: If RPC is really global, expected count of registrations
 101             // of this method is really low.
 102             RpcRouter<T> potentialRouter = getRpcRouter(type);
 103             checkState(potentialRouter.getDefaultService() == null,
 104                         "Default service for routed RPC already registered.");
 105             return potentialRouter.registerDefaultService(implementation);
 106         } catch (RpcIsNotRoutedException e) {
 107             // NOOP - we could safely continue, since RPC is not routed
 108             // so we fallback to global routing.
 109             LOG.debug("RPC is not routed. Using global registration.",e);
 110         }
 111         T publicProxy = getRpcService(type);
 112         RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy);
 113         checkState(currentDelegate == null, "Rpc service is already registered");
 114         LOG.debug("Registering {} as global implementation of {} in {}", implementation, type.getSimpleName(), this);
 115         RuntimeCodeHelper.setDelegate(publicProxy, implementation);
 116         notifyGlobalRpcAdded(type);
 117         return new RpcProxyRegistration<T>(type, implementation, this);
 118     }
 119
 120     @SuppressWarnings("unchecked")
 121     @Override
 122     public final <T extends RpcService> T getRpcService(final Class<T> type) {
 123         return (T) publicProxies.getUnchecked(type);
 124     }
 125
 126
 127     public <T extends RpcService> RpcRouter<T> getRpcRouter(final Class<T> type) {
 128         try {
 129             final AtomicBoolean created = new AtomicBoolean(false);
 130             @SuppressWarnings( "unchecked")
 131             // LoadingCache is unsuitable for RpcRouter since we need to distinguish
 132             // first creation of RPC Router, so that is why
 133             // we are using normal cache with load API and shared AtomicBoolean
 134             // for this call, which will be set to true if router was created.
 135             RpcRouter<T> router = (RpcRouter<T>) rpcRouters.get(type,new Callable<RpcRouter<?>>() {
 136
 137                 @Override
 138                 public org.opendaylight.controller.sal.binding.api.rpc.RpcRouter<?> call()  {
 139                     RpcRouter<?> router = rpcFactory.getRouterFor(type, name);
 140                     router.registerRouteChangeListener(new RouteChangeForwarder<T>(type));
 141                     LOG.debug("Registering router {} as global implementation of {} in {}", router, type.getSimpleName(), this);
 142                     RuntimeCodeHelper.setDelegate(getRpcService(type), router.getInvocationProxy());
 143                     created.set(true);
 144                     return router;
 145                 }
 146             });
 147             if(created.get()) {
 148                 notifyListenersRoutedCreated(router);
 149             }
 150             return router;
 151         } catch (ExecutionException | UncheckedExecutionException e) {
 152             // We rethrow Runtime Exceptions which were wrapped by
 153             // Execution Exceptions
 154             // otherwise we throw IllegalStateException with original
 155             Throwables.propagateIfPossible(e.getCause());
 156             throw new IllegalStateException("Could not load RPC Router for "+type.getName(),e);
 157         }
 158     }
 159
 160     private void notifyGlobalRpcAdded(final Class<? extends RpcService> type) {
 161         for(ListenerRegistration<GlobalRpcRegistrationListener> listener : globalRpcListeners) {
 162             try {
 163                 listener.getInstance().onGlobalRpcRegistered(type);
 164             } catch (Exception e) {
 165                 LOG.error("Unhandled exception during invoking listener {}", e);
 166             }
 167         }
 168
 169     }
 170
 171     private void notifyListenersRoutedCreated(final RpcRouter<?> router) {
 172
 173         for (ListenerRegistration<RouterInstantiationListener> listener : routerInstantiationListener) {
 174             try {
 175                 listener.getInstance().onRpcRouterCreated(router);
 176             } catch (Exception e) {
 177                 LOG.error("Unhandled exception during invoking listener {}", e);
 178             }
 179         }
 180
 181     }
 182
 183     public ListenerRegistration<RouterInstantiationListener> registerRouterInstantiationListener(
 184             final RouterInstantiationListener listener) {
 185         ListenerRegistration<RouterInstantiationListener> reg = routerInstantiationListener.register(listener);
 186         try {
 187             for (RpcRouter<?> router : rpcRouters.asMap().values()) {
 188                 listener.onRpcRouterCreated(router);
 189             }
 190         } catch (Exception e) {
 191             LOG.error("Unhandled exception during invoking listener {}", e);
 192         }
 193         return reg;
 194     }
 195
 196     @SuppressWarnings("unchecked")
 197     @Override
 198     public <L extends RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> ListenerRegistration<L> registerRouteChangeListener(
 199             final L listener) {
 200         return (ListenerRegistration<L>) routeChangeListeners.register(listener);
 201     }
 202
 203     public RuntimeCodeGenerator getRpcFactory() {
 204         return rpcFactory;
 205     }
 206
 207     public void setRpcFactory(final RuntimeCodeGenerator rpcFactory) {
 208         this.rpcFactory = rpcFactory;
 209     }
 210
 211     public interface RouterInstantiationListener extends EventListener {
 212         void onRpcRouterCreated(RpcRouter<?> router);
 213     }
 214
 215     public ListenerRegistration<GlobalRpcRegistrationListener> registerGlobalRpcRegistrationListener(final GlobalRpcRegistrationListener listener) {
 216         return globalRpcListeners.register(listener);
 217     }
 218
 219     public interface GlobalRpcRegistrationListener extends EventListener {
 220         void onGlobalRpcRegistered(Class<? extends RpcService> cls);
 221         void onGlobalRpcUnregistered(Class<? extends RpcService> cls);
 222
 223     }
 224
 225     private class RouteChangeForwarder<T extends RpcService> implements RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>> {
 226
 227         private final Class<T> type;
 228
 229         public RouteChangeForwarder(final Class<T> type) {
 230             this.type = type;
 231         }
 232
 233         @Override
 234         public void onRouteChange(final RouteChange<Class<? extends BaseIdentity>, InstanceIdentifier<?>> change) {
 235             Map<RpcContextIdentifier, Set<InstanceIdentifier<?>>> announcements = new HashMap<>();
 236             for (Entry<Class<? extends BaseIdentity>, Set<InstanceIdentifier<?>>> entry : change.getAnnouncements()
 237                     .entrySet()) {
 238                 RpcContextIdentifier key = RpcContextIdentifier.contextFor(type, entry.getKey());
 239                 announcements.put(key, entry.getValue());
 240             }
 241             Map<RpcContextIdentifier, Set<InstanceIdentifier<?>>> removals = new HashMap<>();
 242             for (Entry<Class<? extends BaseIdentity>, Set<InstanceIdentifier<?>>> entry : change.getRemovals()
 243                     .entrySet()) {
 244                 RpcContextIdentifier key = RpcContextIdentifier.contextFor(type, entry.getKey());
 245                 removals.put(key, entry.getValue());
 246             }
 247             RouteChange<RpcContextIdentifier, InstanceIdentifier<?>> toPublish = RoutingUtils
 248                     .<RpcContextIdentifier, InstanceIdentifier<?>> change(announcements, removals);
 249             for (ListenerRegistration<RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> listener : routeChangeListeners) {
 250                 try {
 251                     listener.getInstance().onRouteChange(toPublish);
 252                 } catch (Exception e) {
 253                     LOG.error("Unhandled exception during invoking listener",listener.getInstance(),e);
 254                 }
 255             }
 256         }
 257     }
 258
 259     public static class RpcProxyRegistration<T extends RpcService> extends AbstractObjectRegistration<T> implements RpcRegistration<T> {
 260
 261         private final Class<T> serviceType;
 262         private RpcProviderRegistryImpl registry;
 263
 264         public RpcProxyRegistration(final Class<T> type, final T service, final RpcProviderRegistryImpl registry) {
 265             super(service);
 266             this.serviceType = type;
 267             this.registry =  registry;
 268         }
 269
 270         @Override
 271         public Class<T> getServiceType() {
 272             return serviceType;
 273         }
 274
 275         @Override
 276         protected void removeRegistration() {
 277             if (registry != null) {
 278                 T publicProxy = registry.getRpcService(serviceType);
 279                 RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy);
 280                 if (currentDelegate == getInstance()) {
 281                     RuntimeCodeHelper.setDelegate(publicProxy, null);
 282                 }
 283                 registry = null;
 284             }
 285         }
 286     }
 287 }