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