X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-dom-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fsal%2Fdom%2Fbroker%2FBrokerImpl.java;h=c15582e95393ee00a81b5b0f9b8432f564fb395c;hp=24b2b4b6daa4e4d47dd40ce85d6b17f0a96be098;hb=3b144c5e66f5e4d14ec881fb553d118e53efd82a;hpb=738d46bed116293e3e42171ad2035ab805b0b2be diff --git a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/BrokerImpl.java b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/BrokerImpl.java index 24b2b4b6da..c15582e953 100644 --- a/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/BrokerImpl.java +++ b/opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/sal/dom/broker/BrokerImpl.java @@ -7,284 +7,217 @@ */ package org.opendaylight.controller.sal.dom.broker; -import java.util.Collection; 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.Future; + +import org.opendaylight.controller.md.sal.common.api.routing.RouteChangeListener; 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.api.RoutedRpcDefaultImplementation; import org.opendaylight.controller.sal.core.api.RpcImplementation; -import org.opendaylight.controller.sal.core.spi.BrokerModule; +import org.opendaylight.controller.sal.core.api.RpcProvisionRegistry; +import org.opendaylight.controller.sal.core.api.RpcRegistrationListener; +import org.opendaylight.controller.sal.core.api.RpcRoutingContext; +import org.opendaylight.controller.sal.dom.broker.spi.RpcRouter; +import org.opendaylight.yangtools.concepts.ListenerRegistration; import org.opendaylight.yangtools.yang.common.QName; import org.opendaylight.yangtools.yang.common.RpcResult; import org.opendaylight.yangtools.yang.data.api.CompositeNode; +import org.opendaylight.yangtools.yang.data.api.InstanceIdentifier; import org.osgi.framework.BundleContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -public class BrokerImpl implements Broker { - private static Logger log = LoggerFactory.getLogger(BrokerImpl.class); +import com.google.common.base.Optional; +import com.google.common.base.Preconditions; +import com.google.common.collect.ClassToInstanceMap; +import com.google.common.collect.ImmutableClassToInstanceMap; +import com.google.common.util.concurrent.ListenableFuture; + +public class BrokerImpl implements Broker, RpcProvisionRegistry, AutoCloseable { + private final static Logger log = LoggerFactory.getLogger(BrokerImpl.class); // Broker Generic Context - private Set sessions = Collections - .synchronizedSet(new HashSet()); - private Set providerSessions = Collections - .synchronizedSet(new HashSet()); - private Set modules = Collections - .synchronizedSet(new HashSet()); - private Map, BrokerModule> serviceProviders = Collections - .synchronizedMap(new HashMap, BrokerModule>()); + private final Set sessions = Collections + .synchronizedSet(new HashSet()); + private final Set providerSessions = Collections + .synchronizedSet(new HashSet()); + + private AutoCloseable deactivator = null; - // RPC Context - private Map rpcImpls = Collections - .synchronizedMap(new HashMap()); + private RpcRouter router = null; - // Implementation specific - private ExecutorService executor; + private final ClassToInstanceMap services; + + public BrokerImpl(final RpcRouter router,final ClassToInstanceMap services) { + this.router = Preconditions.checkNotNull(router, "RPC Router must not be null"); + this.services = ImmutableClassToInstanceMap.copyOf(services); + } - private BundleContext bundleContext; @Override - public ConsumerSession registerConsumer(Consumer consumer,BundleContext ctx) { + public ConsumerSession registerConsumer(final Consumer consumer, + final BundleContext ctx) { checkPredicates(consumer); - log.info("Registering consumer " + consumer); - ConsumerSessionImpl session = newSessionFor(consumer,ctx); + log.trace("Registering consumer {}", consumer); + final ConsumerContextImpl session = newSessionFor(consumer, ctx); consumer.onSessionInitiated(session); sessions.add(session); return session; } @Override - public ProviderSession registerProvider(Provider provider,BundleContext ctx) { + public ProviderSession registerProvider(final Provider provider, + final BundleContext ctx) { checkPredicates(provider); - - ProviderSessionImpl session = newSessionFor(provider,ctx); + final ProviderContextImpl session = newSessionFor(provider, ctx); provider.onSessionInitiated(session); providerSessions.add(session); return session; } - public void addModule(BrokerModule module) { - log.info("Registering broker module " + module); - if (modules.contains(module)) { - log.error("Module already registered"); - throw new IllegalArgumentException("Module already exists."); - } - - Set> provServices = module - .getProvidedServices(); - for (Class serviceType : provServices) { - log.info(" Registering session service implementation: " - + serviceType.getCanonicalName()); - serviceProviders.put(serviceType, module); - } - } - - public T serviceFor(Class service, - ConsumerSessionImpl session) { - BrokerModule prov = serviceProviders.get(service); - if (prov == null) { - log.warn("Service " + service.toString() + " is not supported"); - return null; - } - return prov.getServiceForSession(service, session); + protected Future> invokeRpcAsync(final QName rpc, + final CompositeNode input) { + return router.invokeRpc(rpc, input); } - // RPC Functionality - - private void addRpcImplementation(QName rpcType, - RpcImplementation implementation) { - synchronized (rpcImpls) { - if (rpcImpls.get(rpcType) != null) { - throw new IllegalStateException("Implementation for rpc " - + rpcType + " is already registered."); + // Validation + private void checkPredicates(final Provider prov) { + Preconditions.checkNotNull(prov, "Provider should not be null."); + for (ProviderContextImpl session : providerSessions) { + if (prov.equals(session.getProvider())) { + throw new IllegalStateException("Provider already registered"); } - rpcImpls.put(rpcType, implementation); } - // TODO Add notification for availability of Rpc Implementation + } - private void removeRpcImplementation(QName rpcType, - RpcImplementation implToRemove) { - synchronized (rpcImpls) { - if (implToRemove == rpcImpls.get(rpcType)) { - rpcImpls.remove(rpcType); + private void checkPredicates(final Consumer cons) { + Preconditions.checkNotNull(cons, "Consumer should not be null."); + for (ConsumerContextImpl session : sessions) { + if (cons.equals(session.getConsumer())) { + throw new IllegalStateException("Consumer already registered"); } } - // TODO Add notification for removal of Rpc Implementation } - private Future> invokeRpc(QName rpc, - CompositeNode input) { - RpcImplementation impl = rpcImpls.get(rpc); - // if() - - Callable> call = callableFor(impl, - rpc, input); - Future> result = executor.submit(call); - - return result; + // Private Factory methods + private ConsumerContextImpl newSessionFor(final Consumer provider, + final BundleContext ctx) { + ConsumerContextImpl ret = new ConsumerContextImpl(provider, this); + return ret; } - - // Validation - private void checkPredicates(Provider prov) { - if (prov == null) - throw new IllegalArgumentException("Provider should not be null."); - for (ProviderSessionImpl session : providerSessions) { - if (prov.equals(session.getProvider())) - throw new IllegalStateException("Provider already registered"); - } + private ProviderContextImpl newSessionFor(final Provider provider, + final BundleContext ctx) { + ProviderContextImpl ret = new ProviderContextImpl(provider, this); + return ret; + } + protected void consumerSessionClosed( + final ConsumerContextImpl consumerContextImpl) { + sessions.remove(consumerContextImpl); + providerSessions.remove(consumerContextImpl); } - private void checkPredicates(Consumer cons) { - if (cons == null) - throw new IllegalArgumentException("Consumer should not be null."); - for (ConsumerSessionImpl session : sessions) { - if (cons.equals(session.getConsumer())) - throw new IllegalStateException("Consumer already registered"); + @Override + public void close() throws Exception { + if (deactivator != null) { + deactivator.close(); + deactivator = null; } } - // Private Factory methods - - private ConsumerSessionImpl newSessionFor(Consumer provider, BundleContext ctx) { - return new ConsumerSessionImpl(provider,ctx); + @Override + public RpcRegistration addRpcImplementation(final QName rpcType, + final RpcImplementation implementation) + throws IllegalArgumentException { + return router.addRpcImplementation(rpcType, implementation); } - private ProviderSessionImpl newSessionFor(Provider provider, BundleContext ctx) { - return new ProviderSessionImpl(provider,ctx); + @Override + public RoutedRpcRegistration addRoutedRpcImplementation( + final QName rpcType, final RpcImplementation implementation) { + return router.addRoutedRpcImplementation(rpcType, implementation); } - private void consumerSessionClosed(ConsumerSessionImpl consumerSessionImpl) { - sessions.remove(consumerSessionImpl); - providerSessions.remove(consumerSessionImpl); + @Override + public void setRoutedRpcDefaultDelegate( + final RoutedRpcDefaultImplementation defaultImplementation) { + router.setRoutedRpcDefaultDelegate(defaultImplementation); } - private static Callable> callableFor( - final RpcImplementation implemenation, final QName rpc, - final CompositeNode input) { - - return new Callable>() { - - @Override - public RpcResult call() throws Exception { - return implemenation.invokeRpc(rpc, input); - } - }; + @Override + public ListenerRegistration addRpcRegistrationListener( + final RpcRegistrationListener listener) { + return router.addRpcRegistrationListener(listener); } - - private class ConsumerSessionImpl implements ConsumerSession { - - private final Consumer consumer; - private Map, BrokerService> instantiatedServices = Collections - .synchronizedMap(new HashMap, BrokerService>()); - private boolean closed = false; - - private BundleContext context; - - public Consumer getConsumer() { - return consumer; - } - - public ConsumerSessionImpl(Consumer consumer, BundleContext ctx) { - this.consumer = consumer; - this.context = ctx; - } - - @Override - public Future> rpc(QName rpc, - CompositeNode input) { - return BrokerImpl.this.invokeRpc(rpc, input); - } - - @Override - public T getService(Class service) { - BrokerService potential = instantiatedServices.get(service); - if (potential != null) { - @SuppressWarnings("unchecked") - T ret = (T) potential; - return ret; - } - T ret = BrokerImpl.this.serviceFor(service, this); - if (ret != null) { - instantiatedServices.put(service, ret); - } - return ret; - } + @Override + public > ListenerRegistration registerRouteChangeListener( + final L listener) { + return router.registerRouteChangeListener(listener); + } - @Override - public void close() { - Collection toStop = instantiatedServices.values(); - this.closed = true; - for (BrokerService brokerService : toStop) { - brokerService.closeSession(); - } - BrokerImpl.this.consumerSessionClosed(this); - } + @Override + public Set getSupportedRpcs() { + return router.getSupportedRpcs(); + } - @Override - public boolean isClosed() { - return closed; - } + @Override + public ListenableFuture> invokeRpc( + final QName rpc, final CompositeNode input) { + return router.invokeRpc(rpc, input); + } + /** + * @return the deactivator + */ + public AutoCloseable getDeactivator() { + return deactivator; } - private class ProviderSessionImpl extends ConsumerSessionImpl implements - ProviderSession { + /** + * @param deactivator + * the deactivator to set + */ + public void setDeactivator(final AutoCloseable deactivator) { + this.deactivator = deactivator; + } - private Provider provider; - private Map sessionRpcImpls = Collections.synchronizedMap(new HashMap()); + /** + * @return the router + */ + public RpcRouter getRouter() { + return router; + } - public ProviderSessionImpl(Provider provider, BundleContext ctx) { - super(null,ctx); - this.provider = provider; - } + /** + * @param router + * the router to set + */ + public void setRouter(final RpcRouter router) { + this.router = router; + } - @Override - public void addRpcImplementation(QName rpcType, - RpcImplementation implementation) - throws IllegalArgumentException { - if (rpcType == null) { - throw new IllegalArgumentException("rpcType must not be null"); - } - if (implementation == null) { - throw new IllegalArgumentException( - "Implementation must not be null"); - } - BrokerImpl.this.addRpcImplementation(rpcType, implementation); - sessionRpcImpls.put(rpcType, implementation); - } + protected Optional getGlobalService(final Class service) { + return Optional.fromNullable(services.getInstance(service)); + } - @Override - public void removeRpcImplementation(QName rpcType, - RpcImplementation implToRemove) throws IllegalArgumentException { - RpcImplementation localImpl = rpcImpls.get(rpcType); - if (localImpl != implToRemove) { - throw new IllegalStateException( - "Implementation was not registered in this session"); - } - BrokerImpl.this.removeRpcImplementation(rpcType, implToRemove); - sessionRpcImpls.remove(rpcType); - } + @Override + public ConsumerSession registerConsumer(Consumer cons) { + return registerConsumer(cons,null); + } - public Provider getProvider() { - return this.provider; - } + @Override + public ProviderSession registerProvider(Provider prov) { + return registerProvider(prov,null); } - public void setBundleContext(BundleContext context) { - this.bundleContext = context; - } }