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 com.google.common.cache.CacheBuilder;
  13 import com.google.common.cache.CacheLoader;
  14 import com.google.common.cache.LoadingCache;
  15
  16 import java.util.EventListener;
  17 import java.util.HashMap;
  18 import java.util.Map;
  19 import java.util.Map.Entry;
  20 import java.util.Set;
  21 import java.util.WeakHashMap;
  22
  23 import org.opendaylight.controller.md.sal.common.api.routing.RouteChange;
  24 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener;
  25 import org.opendaylight.controller.md.sal.common.api.routing.RouteChangePublisher;
  26 import org.opendaylight.controller.md.sal.common.impl.routing.RoutingUtils;
  27 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RoutedRpcRegistration;
  28 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration;
  29 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  30 import org.opendaylight.controller.sal.binding.api.rpc.RpcContextIdentifier;
  31 import org.opendaylight.controller.sal.binding.api.rpc.RpcRouter;
  32 import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeGenerator;
  33 import org.opendaylight.controller.sal.binding.codegen.RuntimeCodeHelper;
  34 import org.opendaylight.controller.sal.binding.codegen.impl.SingletonHolder;
  35 import org.opendaylight.yangtools.concepts.AbstractObjectRegistration;
  36 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  37 import org.opendaylight.yangtools.concepts.util.ListenerRegistry;
  38 import org.opendaylight.yangtools.yang.binding.BaseIdentity;
  39 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  40 import org.opendaylight.yangtools.yang.binding.RpcService;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43
  44 public class RpcProviderRegistryImpl implements RpcProviderRegistry, RouteChangePublisher<RpcContextIdentifier, InstanceIdentifier<?>> {
  45
  46     private RuntimeCodeGenerator rpcFactory = SingletonHolder.RPC_GENERATOR_IMPL;
  47
  48     // cache of proxy objects where each value in the map corresponds to a specific RpcService
  49     private final LoadingCache<Class<? extends RpcService>, RpcService> publicProxies = CacheBuilder.newBuilder().weakKeys().
  50             build(new CacheLoader<Class<? extends RpcService>, RpcService>() {
  51                 @Override
  52                 public RpcService load(final Class<? extends RpcService> type) {
  53                     final RpcService proxy = rpcFactory.getDirectProxyFor(type);
  54                     LOG.debug("Created {} as public proxy for {} in {}", proxy, type.getSimpleName(), this);
  55                     return proxy;
  56                 }
  57             });
  58
  59     private final Map<Class<? extends RpcService>, RpcRouter<?>> rpcRouters = new WeakHashMap<>();
  60     private final ListenerRegistry<RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> routeChangeListeners = ListenerRegistry
  61             .create();
  62     private final ListenerRegistry<RouterInstantiationListener> routerInstantiationListener = ListenerRegistry.create();
  63
  64     private final static Logger LOG = LoggerFactory.getLogger(RpcProviderRegistryImpl.class);
  65
  66     private final String name;
  67
  68     private final ListenerRegistry<GlobalRpcRegistrationListener> globalRpcListeners = ListenerRegistry.create();
  69
  70     public String getName() {
  71         return name;
  72     }
  73
  74     public RpcProviderRegistryImpl(final String name) {
  75         super();
  76         this.name = name;
  77     }
  78
  79     @Override
  80     public final <T extends RpcService> RoutedRpcRegistration<T> addRoutedRpcImplementation(final Class<T> type,
  81             final T implementation) throws IllegalStateException {
  82         return getRpcRouter(type).addRoutedRpcImplementation(implementation);
  83     }
  84
  85     @Override
  86     public final <T extends RpcService> RpcRegistration<T> addRpcImplementation(final Class<T> type, final T implementation)
  87             throws IllegalStateException {
  88         @SuppressWarnings("unchecked")
  89         RpcRouter<T> potentialRouter = (RpcRouter<T>) rpcRouters.get(type);
  90         if (potentialRouter != null) {
  91             checkState(potentialRouter.getDefaultService() == null,
  92                     "Default service for routed RPC already registered.");
  93             return potentialRouter.registerDefaultService(implementation);
  94         }
  95         T publicProxy = getRpcService(type);
  96         RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy);
  97         checkState(currentDelegate == null, "Rpc service is already registered");
  98         LOG.debug("Registering {} as global implementation of {} in {}", implementation, type.getSimpleName(), this);
  99         RuntimeCodeHelper.setDelegate(publicProxy, implementation);
 100         notifyGlobalRpcAdded(type);
 101         return new RpcProxyRegistration<T>(type, implementation, this);
 102     }
 103
 104     @SuppressWarnings("unchecked")
 105     @Override
 106     public final <T extends RpcService> T getRpcService(final Class<T> type) {
 107         return (T) publicProxies.getUnchecked(type);
 108     }
 109
 110     @SuppressWarnings({ "unchecked", "rawtypes" })
 111     public <T extends RpcService> RpcRouter<T> getRpcRouter(final Class<T> type) {
 112         RpcRouter<?> potentialRouter = rpcRouters.get(type);
 113         if (potentialRouter != null) {
 114             return (RpcRouter<T>) potentialRouter;
 115         }
 116         synchronized (this) {
 117             /**
 118              * Potential Router could be instantiated by other thread while we
 119              * were waiting for the lock.
 120              */
 121             potentialRouter = rpcRouters.get(type);
 122             if (potentialRouter != null) {
 123                 return (RpcRouter<T>) potentialRouter;
 124             }
 125             RpcRouter<T> router = rpcFactory.getRouterFor(type, name);
 126             router.registerRouteChangeListener(new RouteChangeForwarder(type));
 127             LOG.debug("Registering router {} as global implementation of {} in {}", router, type.getSimpleName(), this);
 128             RuntimeCodeHelper.setDelegate(getRpcService(type), router.getInvocationProxy());
 129             rpcRouters.put(type, router);
 130             notifyListenersRoutedCreated(router);
 131             return router;
 132         }
 133     }
 134
 135     private void notifyGlobalRpcAdded(final Class<? extends RpcService> type) {
 136         for(ListenerRegistration<GlobalRpcRegistrationListener> listener : globalRpcListeners) {
 137             try {
 138                 listener.getInstance().onGlobalRpcRegistered(type);
 139             } catch (Exception e) {
 140                 LOG.error("Unhandled exception during invoking listener {}", e);
 141             }
 142         }
 143
 144     }
 145
 146     private void notifyListenersRoutedCreated(final RpcRouter<?> router) {
 147
 148         for (ListenerRegistration<RouterInstantiationListener> listener : routerInstantiationListener) {
 149             try {
 150                 listener.getInstance().onRpcRouterCreated(router);
 151             } catch (Exception e) {
 152                 LOG.error("Unhandled exception during invoking listener {}", e);
 153             }
 154         }
 155
 156     }
 157
 158     public ListenerRegistration<RouterInstantiationListener> registerRouterInstantiationListener(
 159             final RouterInstantiationListener listener) {
 160         ListenerRegistration<RouterInstantiationListener> reg = routerInstantiationListener.register(listener);
 161         try {
 162             for (RpcRouter<?> router : rpcRouters.values()) {
 163                 listener.onRpcRouterCreated(router);
 164             }
 165         } catch (Exception e) {
 166             LOG.error("Unhandled exception during invoking listener {}", e);
 167         }
 168         return reg;
 169     }
 170
 171     @SuppressWarnings("unchecked")
 172     @Override
 173     public <L extends RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> ListenerRegistration<L> registerRouteChangeListener(
 174             final L listener) {
 175         return (ListenerRegistration<L>) routeChangeListeners.register(listener);
 176     }
 177
 178     public RuntimeCodeGenerator getRpcFactory() {
 179         return rpcFactory;
 180     }
 181
 182     public void setRpcFactory(final RuntimeCodeGenerator rpcFactory) {
 183         this.rpcFactory = rpcFactory;
 184     }
 185
 186     public interface RouterInstantiationListener extends EventListener {
 187         void onRpcRouterCreated(RpcRouter<?> router);
 188     }
 189
 190     public ListenerRegistration<GlobalRpcRegistrationListener> registerGlobalRpcRegistrationListener(final GlobalRpcRegistrationListener listener) {
 191         return globalRpcListeners.register(listener);
 192     }
 193
 194     public interface GlobalRpcRegistrationListener extends EventListener {
 195         void onGlobalRpcRegistered(Class<? extends RpcService> cls);
 196         void onGlobalRpcUnregistered(Class<? extends RpcService> cls);
 197
 198     }
 199
 200     private class RouteChangeForwarder<T extends RpcService> implements RouteChangeListener<Class<? extends BaseIdentity>, InstanceIdentifier<?>> {
 201
 202         private final Class<T> type;
 203
 204         public RouteChangeForwarder(final Class<T> type) {
 205             this.type = type;
 206         }
 207
 208         @Override
 209         public void onRouteChange(final RouteChange<Class<? extends BaseIdentity>, InstanceIdentifier<?>> change) {
 210             Map<RpcContextIdentifier, Set<InstanceIdentifier<?>>> announcements = new HashMap<>();
 211             for (Entry<Class<? extends BaseIdentity>, Set<InstanceIdentifier<?>>> entry : change.getAnnouncements()
 212                     .entrySet()) {
 213                 RpcContextIdentifier key = RpcContextIdentifier.contextFor(type, entry.getKey());
 214                 announcements.put(key, entry.getValue());
 215             }
 216             Map<RpcContextIdentifier, Set<InstanceIdentifier<?>>> removals = new HashMap<>();
 217             for (Entry<Class<? extends BaseIdentity>, Set<InstanceIdentifier<?>>> entry : change.getRemovals()
 218                     .entrySet()) {
 219                 RpcContextIdentifier key = RpcContextIdentifier.contextFor(type, entry.getKey());
 220                 removals.put(key, entry.getValue());
 221             }
 222             RouteChange<RpcContextIdentifier, InstanceIdentifier<?>> toPublish = RoutingUtils
 223                     .<RpcContextIdentifier, InstanceIdentifier<?>> change(announcements, removals);
 224             for (ListenerRegistration<RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> listener : routeChangeListeners) {
 225                 try {
 226                     listener.getInstance().onRouteChange(toPublish);
 227                 } catch (Exception e) {
 228                     e.printStackTrace();
 229                 }
 230             }
 231         }
 232     }
 233
 234     public static class RpcProxyRegistration<T extends RpcService> extends AbstractObjectRegistration<T> implements RpcRegistration<T> {
 235
 236         private final Class<T> serviceType;
 237         private RpcProviderRegistryImpl registry;
 238
 239         public RpcProxyRegistration(final Class<T> type, final T service, final RpcProviderRegistryImpl registry) {
 240             super(service);
 241             this.serviceType = type;
 242             this.registry =  registry;
 243         }
 244
 245         @Override
 246         public Class<T> getServiceType() {
 247             return serviceType;
 248         }
 249
 250         @Override
 251         protected void removeRegistration() {
 252             if (registry != null) {
 253                 T publicProxy = registry.getRpcService(serviceType);
 254                 RpcService currentDelegate = RuntimeCodeHelper.getDelegate(publicProxy);
 255                 if (currentDelegate == getInstance()) {
 256                     RuntimeCodeHelper.setDelegate(publicProxy, null);
 257                 }
 258                 registry = null;
 259             }
 260         }
 261     }
 262 }