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