import org.opendaylight.controller.sal.binding.api.BindingAwareConsumer
import org.opendaylight.controller.sal.binding.api.BindingAwareProvider
-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
-import static extension org.opendaylight.controller.sal.binding.codegen.RuntimeCodeHelper.*
-
import org.opendaylight.controller.sal.binding.api.NotificationProviderService
-import org.osgi.framework.ServiceRegistration
-import static org.opendaylight.controller.sal.binding.impl.osgi.Constants.*
-import static extension org.opendaylight.controller.sal.binding.impl.osgi.PropertiesUtils.*
-import org.opendaylight.controller.sal.binding.api.NotificationService
import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.ProviderContext
-
-import org.slf4j.LoggerFactory
-import org.opendaylight.controller.sal.binding.codegen.impl.RuntimeCodeGenerator
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier
-import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RoutedRpcRegistration
-import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.RpcRegistration
-import org.opendaylight.controller.sal.binding.api.data.DataProviderService
-import org.opendaylight.controller.sal.binding.api.data.DataBrokerService
-import org.opendaylight.controller.sal.binding.spi.RpcRouter
-import java.util.concurrent.ConcurrentHashMap
-import static com.google.common.base.Preconditions.*
-import org.opendaylight.yangtools.concepts.AbstractObjectRegistration
-import org.opendaylight.yangtools.yang.binding.BaseIdentity
-import com.google.common.collect.Multimap
-import com.google.common.collect.HashMultimap
-import static org.opendaylight.controller.sal.binding.impl.util.ClassLoaderUtils.*
-import java.util.concurrent.Executors
-import java.util.Collections
import org.opendaylight.yangtools.yang.binding.DataObject
-import org.opendaylight.controller.sal.binding.impl.connect.dom.ConnectorActivator
+import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener
+import org.opendaylight.controller.sal.binding.spi.RpcContextIdentifier
+import org.opendaylight.controller.sal.binding.api.data.DataProviderService
+import org.slf4j.LoggerFactory
-class BindingAwareBrokerImpl implements BindingAwareBroker, AutoCloseable {
+class BindingAwareBrokerImpl extends RpcProviderRegistryImpl implements BindingAwareBroker, AutoCloseable {
private static val log = LoggerFactory.getLogger(BindingAwareBrokerImpl)
-
private InstanceIdentifier<? extends DataObject> root = InstanceIdentifier.builder().toInstance();
- private val clsPool = ClassPool.getDefault()
- private 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 ConcurrentHashMap();
-
@Property
- private var NotificationBrokerImpl notifyBroker
-
+ private var NotificationProviderService notifyBroker
+
@Property
- private var DataBrokerImpl dataBroker
-
+ private var DataProviderService dataBroker
+
@Property
var BundleContext brokerBundleContext
-
- ServiceRegistration<NotificationProviderService> notifyProviderRegistration
-
- ServiceRegistration<NotificationService> notifyConsumerRegistration
-
- ServiceRegistration<DataProviderService> dataProviderRegistration
-
- ServiceRegistration<DataBrokerService> dataConsumerRegistration
-
- ConnectorActivator connectorActivator
-
-
- public new(BundleContext bundleContext) {
+
+ public new(String name,BundleContext bundleContext) {
+ super(name);
_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");
- notifyBroker = new NotificationBrokerImpl(executor);
- notifyBroker.invokerFactory = generator.invokerFactory;
-
- log.info("Starting MD-SAL: Binding Aware Data Broker");
- dataBroker = new DataBrokerImpl();
- dataBroker.executor = executor;
-
- val brokerProperties = newProperties();
-
-
- log.info("Starting MD-SAL: Binding Aware Data Broker");
- notifyProviderRegistration = brokerBundleContext.registerService(NotificationProviderService, notifyBroker,
- brokerProperties)
- notifyConsumerRegistration = brokerBundleContext.registerService(NotificationService, notifyBroker, brokerProperties)
- dataProviderRegistration = brokerBundleContext.registerService(DataProviderService, dataBroker, brokerProperties)
- dataConsumerRegistration = brokerBundleContext.registerService(DataBrokerService, dataBroker, brokerProperties)
-
- connectorActivator = new ConnectorActivator(dataBroker,brokerBundleContext);
- connectorActivator.start();
- 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)
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
- }
- 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
- }
-
- /**
- * Registers RPC Implementation
- *
- */
- def <T extends RpcService> registerRpcImplementation(Class<T> type, T service, OsgiProviderContext context,
- Hashtable<String, String> properties) {
- val proxy = getManagedDirectProxy(type)
- checkState(proxy.delegate === null, "The Service for type %s is already registered", type)
-
- val osgiReg = context.bundleContext.registerService(type, service, properties);
- proxy.delegate = service;
- return new RpcServiceRegistrationImpl<T>(type, service, osgiReg,this);
- }
-
- def <T extends RpcService> RoutedRpcRegistration<T> registerRoutedRpcImplementation(Class<T> type, T service,
- OsgiProviderContext context) {
- val router = resolveRpcRouter(type);
- checkState(router !== null)
- return new RoutedRpcRegistrationImpl<T>(service, router, this)
- }
-
- 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
- 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)
- }
+ override <L extends RouteChangeListener<RpcContextIdentifier, InstanceIdentifier<?>>> registerRouteChangeListener(L listener) {
+ super.<L>registerRouteChangeListener(listener)
}
- protected def <T extends RpcService> void unregisterRpcService(RpcServiceRegistrationImpl<T> registration) {
-
- val type = registration.serviceType;
+ override close() throws Exception {
- 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> {
-
- val ServiceRegistration<T> osgiRegistration;
- private var BindingAwareBrokerImpl broker;
-
- @Property
- val Class<T> serviceType;
-
- public new(Class<T> type, T service, ServiceRegistration<T> osgiReg,BindingAwareBrokerImpl broker) {
- super(service);
- this._serviceType = type;
- this.osgiRegistration = osgiReg;
- this.broker= broker;
- }
-
- override protected removeRegistration() {
- broker.unregisterRpcService(this);
- broker = null;
- }
-
-}
+}
\ No newline at end of file
Code Review / controller.git / blobdiff