Fixed bug in cross-broker RPC routing
[controller.git] / opendaylight / md-sal / sal-binding-broker / src / main / java / org / opendaylight / controller / sal / binding / impl / BindingAwareBrokerImpl.xtend
index 9381a5a..8d3545f 100644 (file)
@@ -13,7 +13,6 @@ import org.opendaylight.yangtools.yang.binding.RpcService
 import javassist.ClassPool
 import org.osgi.framework.BundleContext
 import java.util.Map
-import java.util.HashMap
 import javassist.LoaderClassPath
 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker
 import java.util.Hashtable
@@ -49,29 +48,14 @@ import java.util.concurrent.Callable
 import java.util.WeakHashMap
 import javax.annotation.concurrent.GuardedBy
 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry
+import org.opendaylight.yangtools.concepts.ListenerRegistration
+import org.opendaylight.yangtools.concepts.util.ListenerRegistry
 
-class BindingAwareBrokerImpl implements BindingAwareBroker, RpcProviderRegistry, AutoCloseable {
+class BindingAwareBrokerImpl extends RpcProviderRegistryImpl implements BindingAwareBroker, AutoCloseable {
     private static val log = LoggerFactory.getLogger(BindingAwareBrokerImpl)
 
     private InstanceIdentifier<? extends DataObject> root = InstanceIdentifier.builder().toInstance();
 
-    private static val clsPool = ClassPool.getDefault()
-    public static var RuntimeCodeGenerator generator;
-
-    /**
-     * Map of all Managed Direct Proxies
-     * 
-     */
-    private val Map<Class<? extends RpcService>, RpcProxyContext> managedProxies = new ConcurrentHashMap();
-
-    /**
-     * 
-     * Map of all available Rpc Routers
-     * 
-     * 
-     */
-    private val Map<Class<? extends RpcService>, RpcRouter<? extends RpcService>> rpcRouters = new WeakHashMap();
-
     @Property
     private var NotificationProviderService notifyBroker
 
@@ -81,43 +65,15 @@ class BindingAwareBrokerImpl implements BindingAwareBroker, RpcProviderRegistry,
     @Property
     var BundleContext brokerBundleContext
 
-    ServiceRegistration<NotificationProviderService> notifyProviderRegistration
-
-    ServiceRegistration<NotificationService> notifyConsumerRegistration
-
-    ServiceRegistration<DataProviderService> dataProviderRegistration
-
-    ServiceRegistration<DataBrokerService> dataConsumerRegistration
-
-    private val proxyGenerationLock = new ReentrantLock;
-
-    private val routerGenerationLock = new ReentrantLock;
-
     public new(BundleContext bundleContext) {
         _brokerBundleContext = bundleContext;
     }
 
     def start() {
         log.info("Starting MD-SAL: Binding Aware Broker");
-        initGenerator();
-
-        val executor = Executors.newCachedThreadPool;
-
-        // Initialization of notificationBroker
-        log.info("Starting MD-SAL: Binding Aware Notification Broker");
-
-        log.info("Starting MD-SAL: Binding Aware Data Broker");
-
-        log.info("Starting MD-SAL: Binding Aware Data Broker");
-        log.info("MD-SAL: Binding Aware Broker Started");
     }
 
-    def initGenerator() {
 
-        // YANG Binding Class Loader
-        clsPool.appendClassPath(new LoaderClassPath(RpcService.classLoader));
-        generator = new RuntimeCodeGenerator(clsPool);
-    }
 
     override registerConsumer(BindingAwareConsumer consumer, BundleContext bundleCtx) {
         val ctx = consumer.createContext(bundleCtx)
@@ -139,247 +95,9 @@ class BindingAwareBrokerImpl implements BindingAwareBroker, RpcProviderRegistry,
     private def createContext(BindingAwareProvider provider, BundleContext providerCtx) {
         new OsgiProviderContext(providerCtx, this)
     }
-
-    /**
-     * Returns a Managed Direct Proxy for supplied class
-     * 
-     * Managed direct proxy is a generated proxy class conforming to the supplied interface
-     * which delegates all calls to the backing delegate.
-     * 
-     * Proxy does not do any validation, null pointer checks or modifies data in any way, it
-     * is only use to avoid exposing direct references to backing implementation of service.
-     * 
-     * If proxy class does not exist for supplied service class it will be generated automatically.
-     */
-    private def <T extends RpcService> getManagedDirectProxy(Class<T> service) {
-        var RpcProxyContext existing = null
-
-        if ((existing = managedProxies.get(service)) != null) {
-            return existing.proxy
-        }
-        return withLock(proxyGenerationLock) [ |
-            val maybeProxy = managedProxies.get(service);
-            if (maybeProxy !== null) {
-                return maybeProxy.proxy;
-            }
-            
-            
-            val proxyInstance = generator.getDirectProxyFor(service)
-            val rpcProxyCtx = new RpcProxyContext(proxyInstance.class)
-            val properties = new Hashtable<String, String>()
-            rpcProxyCtx.proxy = proxyInstance as RpcService
-            properties.salServiceType = SAL_SERVICE_TYPE_CONSUMER_PROXY
-            rpcProxyCtx.registration = brokerBundleContext.registerService(service, rpcProxyCtx.proxy as T, properties)
-            managedProxies.put(service, rpcProxyCtx)
-            return rpcProxyCtx.proxy
-        ]
-    }
-
-    private static def <T> T withLock(ReentrantLock lock, Callable<T> method) {
-        try {
-            lock.lock();
-            val ret = method.call;
-            return ret;
-        } finally {
-            lock.unlock();
-        }
-    }
-
-    /**
-     * Registers RPC Implementation
-     * 
-     */
-    override <T extends RpcService> addRpcImplementation(Class<T> type, T service) {
-        checkNotNull(type, "Service type should not be null")
-        checkNotNull(service, "Service type should not be null")
-        
-        val proxy = getManagedDirectProxy(type)
-        checkState(proxy.delegate === null, "The Service for type %s is already registered", type)
-
-        proxy.delegate = service;
-        return new RpcServiceRegistrationImpl<T>(type, service, this);
-    }
-
-    override <T extends RpcService> RoutedRpcRegistration<T> addRoutedRpcImplementation(Class<T> type, T service) {
-        checkNotNull(type, "Service type should not be null")
-        checkNotNull(service, "Service type should not be null")
-        
-        val router = resolveRpcRouter(type);
-        checkState(router !== null)
-        return new RoutedRpcRegistrationImpl<T>(service, router, this)
-    }
     
-    override <T extends RpcService> getRpcService(Class<T> service) {
-        checkNotNull(service, "Service should not be null");
-        return getManagedDirectProxy(service) as T;
-    }
-
-    private def <T extends RpcService> RpcRouter<T> resolveRpcRouter(Class<T> type) {
-
-        val router = rpcRouters.get(type);
-        if (router !== null) {
-            return router as RpcRouter<T>;
-        }
-
-        // We created Router
-        return withLock(routerGenerationLock) [ |
-            val maybeRouter = rpcRouters.get(type);
-            if (maybeRouter !== null) {
-                return maybeRouter as RpcRouter<T>;
-            }
-            
-            val newRouter = generator.getRouterFor(type);
-            checkState(newRouter !== null);
-            rpcRouters.put(type, newRouter);
-            // We create / update Direct Proxy for router
-            val proxy = getManagedDirectProxy(type);
-            proxy.delegate = newRouter.invocationProxy
-            return newRouter;
-        ]
-
-    }
-
-    protected def <T extends RpcService> void registerPath(RoutedRpcRegistrationImpl<T> registration,
-        Class<? extends BaseIdentity> context, InstanceIdentifier<? extends Object> path) {
-
-        val router = registration.router;
-        val paths = registration.registeredPaths;
-
-        val routingTable = router.getRoutingTable(context)
-        checkState(routingTable != null);
-
-        // Updating internal structure of registration
-        routingTable.updateRoute(path, registration.instance)
-
-        // Update routing table / send announce to message bus
-        val success = paths.put(context, path);
-    }
-
-    protected def <T extends RpcService> void unregisterPath(RoutedRpcRegistrationImpl<T> registration,
-        Class<? extends BaseIdentity> context, InstanceIdentifier<? extends Object> path) {
-
-        val router = registration.router;
-        val paths = registration.registeredPaths;
-
-        val routingTable = router.getRoutingTable(context)
-        checkState(routingTable != null);
-
-        // Updating internal structure of registration
-        val target = routingTable.getRoute(path)
-        checkState(target === registration.instance)
-        routingTable.removeRoute(path)
-        checkState(paths.remove(context, path));
-    }
-
-    protected def <T extends RpcService> void unregisterRoutedRpcService(RoutedRpcRegistrationImpl<T> registration) {
-
-        val router = registration.router;
-        val paths = registration.registeredPaths;
-
-        for (ctxMap : registration.registeredPaths.entries) {
-            val context = ctxMap.key
-            val routingTable = router.getRoutingTable(context)
-            val path = ctxMap.value
-            routingTable.removeRoute(path)
-        }
-    }
-
-    protected def <T extends RpcService> void unregisterRpcService(RpcServiceRegistrationImpl<T> registration) {
-
-        val type = registration.serviceType;
-
-        val proxy = managedProxies.get(type);
-        if (proxy.proxy.delegate === registration.instance) {
-            proxy.proxy.delegate = null;
-        }
-    }
-
-    def createDelegate(Class<? extends RpcService> type) {
-        getManagedDirectProxy(type);
-    }
-
-    def getRpcRouters() {
-        return Collections.unmodifiableMap(rpcRouters);
-    }
-
-    override close() {
-        dataConsumerRegistration.unregister()
-        dataProviderRegistration.unregister()
-        notifyConsumerRegistration.unregister()
-        notifyProviderRegistration.unregister()
-    }
-
-}
-
-class RoutedRpcRegistrationImpl<T extends RpcService> extends AbstractObjectRegistration<T> implements RoutedRpcRegistration<T> {
-
-    @Property
-    private val BindingAwareBrokerImpl broker;
-
-    @Property
-    private val RpcRouter<T> router;
-
-    @Property
-    private val Multimap<Class<? extends BaseIdentity>, InstanceIdentifier<?>> registeredPaths = HashMultimap.create();
-
-    private var closed = false;
-
-    new(T instance, RpcRouter<T> backingRouter, BindingAwareBrokerImpl broker) {
-        super(instance)
-        _router = backingRouter;
-        _broker = broker;
-    }
-
-    override protected removeRegistration() {
-        closed = true
-        broker.unregisterRoutedRpcService(this)
-    }
-
-    override registerInstance(Class<? extends BaseIdentity> context, InstanceIdentifier<? extends Object> instance) {
-        registerPath(context, instance);
-    }
-
-    override unregisterInstance(Class<? extends BaseIdentity> context, InstanceIdentifier<? extends Object> instance) {
-        unregisterPath(context, instance);
-    }
-
-    override registerPath(Class<? extends BaseIdentity> context, InstanceIdentifier<? extends Object> path) {
-        checkClosed()
-        broker.registerPath(this, context, path);
-    }
-
-    override unregisterPath(Class<? extends BaseIdentity> context, InstanceIdentifier<? extends Object> path) {
-        checkClosed()
-        broker.unregisterPath(this, context, path);
-    }
-
-    override getServiceType() {
-        return router.serviceType;
-    }
-
-    private def checkClosed() {
-        if (closed)
-            throw new IllegalStateException("Registration was closed.");
-    }
-
-}
-
-class RpcServiceRegistrationImpl<T extends RpcService> extends AbstractObjectRegistration<T> implements RpcRegistration<T> {
-
-    private var BindingAwareBrokerImpl broker;
-
-    @Property
-    val Class<T> serviceType;
-
-    public new(Class<T> type, T service, BindingAwareBrokerImpl broker) {
-        super(service);
-        this._serviceType = type;
-        this.broker = broker;
-    }
-
-    override protected removeRegistration() {
-        broker.unregisterRpcService(this);
-        broker = null;
+    override close() throws Exception {
+        
     }
 
-}
+}
\ No newline at end of file

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.