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