X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-dom-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fsal%2Fdom%2Fbroker%2FBrokerImpl.xtend;h=64de8683d1d1a8a5c3854aa5f55e6b907852c448;hb=43d7c8702fc7a89ca5acdeefb4696c91b2963b38;hp=855ad9bd328d19c79b30e09ba4e485e0d1918044;hpb=4f76ea30ad49331ca38ce63925b3fabf8e769731;p=controller.git diff --git a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/BrokerImpl.xtend b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/BrokerImpl.xtend index 855ad9bd32..64de8683d1 100644 --- a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/BrokerImpl.xtend +++ b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/BrokerImpl.xtend @@ -8,38 +8,36 @@ package org.opendaylight.controller.sal.dom.broker; import java.util.Collections -import java.util.HashMap import java.util.HashSet -import java.util.Map import java.util.Set import java.util.concurrent.Callable import java.util.concurrent.ExecutorService import java.util.concurrent.Executors import java.util.concurrent.Future import org.opendaylight.controller.sal.core.api.Broker -import org.opendaylight.controller.sal.core.api.BrokerService import org.opendaylight.controller.sal.core.api.Consumer import org.opendaylight.controller.sal.core.api.Provider -import org.opendaylight.controller.sal.core.spi.BrokerModule import org.opendaylight.yangtools.yang.common.QName import org.opendaylight.yangtools.yang.common.RpcResult import org.opendaylight.yangtools.yang.data.api.CompositeNode import org.osgi.framework.BundleContext import org.slf4j.LoggerFactory import org.opendaylight.controller.sal.dom.broker.spi.RpcRouter -import org.opendaylight.yangtools.concepts.ListenerRegistration import org.opendaylight.controller.sal.core.api.RpcRegistrationListener - -public class BrokerImpl implements Broker { +import org.opendaylight.controller.sal.core.api.RpcProvisionRegistry +import org.opendaylight.controller.sal.core.api.RpcImplementation +import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener +import org.opendaylight.controller.sal.core.api.RpcRoutingContext +import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier +import org.opendaylight.controller.sal.core.api.RoutedRpcDefaultImplementation + +public class BrokerImpl implements Broker, RpcProvisionRegistry, AutoCloseable { private static val log = LoggerFactory.getLogger(BrokerImpl); // Broker Generic Context private val Set sessions = Collections.synchronizedSet(new HashSet()); private val Set providerSessions = Collections.synchronizedSet( new HashSet()); - private val Set modules = Collections.synchronizedSet(new HashSet()); - private val Map, BrokerModule> serviceProviders = Collections. - synchronizedMap(new HashMap, BrokerModule>()); // Implementation specific @Property @@ -47,12 +45,15 @@ public class BrokerImpl implements Broker { @Property private var BundleContext bundleContext; + @Property + private var AutoCloseable deactivator; + @Property private var RpcRouter router; override registerConsumer(Consumer consumer, BundleContext ctx) { checkPredicates(consumer); - log.info("Registering consumer " + consumer); + log.trace("Registering consumer " + consumer); val session = newSessionFor(consumer, ctx); consumer.onSessionInitiated(session); sessions.add(session); @@ -68,50 +69,27 @@ public class BrokerImpl implements Broker { return session; } - public def addModule(BrokerModule module) { - log.info("Registering broker module " + module); - if(modules.contains(module)) { - log.error("Module already registered"); - throw new IllegalArgumentException("Module already exists."); - } - - val provServices = module.getProvidedServices(); - for (Class serviceType : provServices) { - log.info(" Registering session service implementation: " + serviceType.getCanonicalName()); - serviceProviders.put(serviceType, module); - } - } - - public def T serviceFor(Class service, ConsumerContextImpl session) { - val prov = serviceProviders.get(service); - if(prov == null) { - log.warn("Service " + service.toString() + " is not supported"); - return null; - } - return prov.getServiceForSession(service, session); - } - - protected def Future> invokeRpc(QName rpc, CompositeNode input) { + protected def Future> invokeRpcAsync(QName rpc, CompositeNode input) { val result = executor.submit([|router.invokeRpc(rpc, input)] as Callable>); return result; } // Validation private def void checkPredicates(Provider prov) { - if(prov == null) + if (prov == null) throw new IllegalArgumentException("Provider should not be null."); for (ProviderContextImpl session : providerSessions) { - if(prov.equals(session.getProvider())) + if (prov.equals(session.getProvider())) throw new IllegalStateException("Provider already registered"); } } private def void checkPredicates(Consumer cons) { - if(cons == null) + if (cons == null) throw new IllegalArgumentException("Consumer should not be null."); for (ConsumerContextImpl session : sessions) { - if(cons.equals(session.getConsumer())) + if (cons.equals(session.getConsumer())) throw new IllegalStateException("Consumer already registered"); } } @@ -133,4 +111,37 @@ public class BrokerImpl implements Broker { sessions.remove(consumerContextImpl); providerSessions.remove(consumerContextImpl); } + + override close() throws Exception { + deactivator?.close(); + } + + override addRpcImplementation(QName rpcType, RpcImplementation implementation) throws IllegalArgumentException { + router.addRpcImplementation(rpcType,implementation); + } + + override addRoutedRpcImplementation(QName rpcType, RpcImplementation implementation) { + router.addRoutedRpcImplementation(rpcType,implementation); + } + + override setRoutedRpcDefaultDelegate(RoutedRpcDefaultImplementation defaultImplementation) { + router.setRoutedRpcDefaultDelegate(defaultImplementation); + } + + override addRpcRegistrationListener(RpcRegistrationListener listener) { + return router.addRpcRegistrationListener(listener); + } + + override > registerRouteChangeListener(L listener) { + return router.registerRouteChangeListener(listener); + } + + override invokeRpc(QName rpc,CompositeNode input){ + return router.invokeRpc(rpc,input) + } + + override getSupportedRpcs() { + return router.getSupportedRpcs(); + } + }