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.java;h=e4bd0343cfcf62999d3becca52177eaf8d366bcd;hb=9ba0d83d16dbf7122ab4aeade8e630fe407d9fc3;hp=24b2b4b6daa4e4d47dd40ce85d6b17f0a96be098;hpb=738d46bed116293e3e42171ad2035ab805b0b2be;p=controller.git 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..e4bd0343cf 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,205 @@ */ package org.opendaylight.controller.sal.dom.broker; -import java.util.Collection; +import com.google.common.base.Preconditions; +import com.google.common.util.concurrent.ListenableFuture; + 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.RpcImplementation; -import org.opendaylight.controller.sal.core.spi.BrokerModule; +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.osgi.framework.BundleContext; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.opendaylight.controller.sal.dom.broker.spi.RpcRouter; +import org.opendaylight.controller.sal.core.api.RpcRegistrationListener; +import org.opendaylight.controller.sal.core.api.RpcProvisionRegistry; +import org.opendaylight.controller.sal.core.api.RpcImplementation; +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 { - private static Logger log = LoggerFactory.getLogger(BrokerImpl.class); +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()); - // RPC Context - private Map rpcImpls = Collections - .synchronizedMap(new HashMap()); + private BundleContext bundleContext = null; - // Implementation specific - private ExecutorService executor; + private AutoCloseable deactivator = null; - private BundleContext bundleContext; + private RpcRouter router = null; @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); - } - - // 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."); - } - 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); - } - } - // TODO Add notification for removal of Rpc Implementation + protected Future> invokeRpcAsync(final QName rpc, + final CompositeNode input) { + return router.invokeRpc(rpc, input); } - 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; - } - // Validation - - private void checkPredicates(Provider prov) { - if (prov == null) - throw new IllegalArgumentException("Provider should not be null."); - for (ProviderSessionImpl session : providerSessions) { + 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"); } } - private void checkPredicates(Consumer cons) { - if (cons == null) - throw new IllegalArgumentException("Consumer should not be null."); - for (ConsumerSessionImpl session : sessions) { + 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"); } } // Private Factory methods - - private ConsumerSessionImpl newSessionFor(Consumer provider, BundleContext ctx) { - return new ConsumerSessionImpl(provider,ctx); + private ConsumerContextImpl newSessionFor(final Consumer provider, + final BundleContext ctx) { + ConsumerContextImpl ret = new ConsumerContextImpl(provider, ctx); + ret.setBroker(this); + return ret; } - private ProviderSessionImpl newSessionFor(Provider provider, BundleContext ctx) { - return new ProviderSessionImpl(provider,ctx); + private ProviderContextImpl newSessionFor(final Provider provider, + final BundleContext ctx) { + ProviderContextImpl ret = new ProviderContextImpl(provider, ctx); + ret.setBroker(this); + return ret; } - private void consumerSessionClosed(ConsumerSessionImpl consumerSessionImpl) { - sessions.remove(consumerSessionImpl); - providerSessions.remove(consumerSessionImpl); + protected void consumerSessionClosed( + final ConsumerContextImpl consumerContextImpl) { + sessions.remove(consumerContextImpl); + providerSessions.remove(consumerContextImpl); } - 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); - } - }; - } - - 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 void close() throws Exception { + if (deactivator != null) { + deactivator.close(); + deactivator = null; } + } - @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 RpcRegistration addRpcImplementation(final QName rpcType, + final RpcImplementation implementation) + throws IllegalArgumentException { + return router.addRpcImplementation(rpcType, implementation); + } - @Override - public void close() { - Collection toStop = instantiatedServices.values(); - this.closed = true; - for (BrokerService brokerService : toStop) { - brokerService.closeSession(); - } - BrokerImpl.this.consumerSessionClosed(this); - } + @Override + public RoutedRpcRegistration addRoutedRpcImplementation( + final QName rpcType, final RpcImplementation implementation) { + return router.addRoutedRpcImplementation(rpcType, implementation); + } - @Override - public boolean isClosed() { - return closed; - } + @Override + public void setRoutedRpcDefaultDelegate( + final RoutedRpcDefaultImplementation defaultImplementation) { + router.setRoutedRpcDefaultDelegate(defaultImplementation); + } + @Override + public ListenerRegistration addRpcRegistrationListener( + final RpcRegistrationListener listener) { + return router.addRpcRegistrationListener(listener); } - private class ProviderSessionImpl extends ConsumerSessionImpl implements - ProviderSession { + @Override + public > ListenerRegistration registerRouteChangeListener( + final L listener) { + return router.registerRouteChangeListener(listener); + } - private Provider provider; - private Map sessionRpcImpls = Collections.synchronizedMap(new HashMap()); + @Override + public Set getSupportedRpcs() { + return router.getSupportedRpcs(); + } - public ProviderSessionImpl(Provider provider, BundleContext ctx) { - super(null,ctx); - this.provider = provider; - } + @Override + public ListenableFuture> invokeRpc( + final QName rpc, final CompositeNode input) { + return router.invokeRpc(rpc, input); + } - @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); - } + /** + * @return the bundleContext + */ + public BundleContext getBundleContext() { + return bundleContext; + } - @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"); - } + /** + * @param bundleContext + * the bundleContext to set + */ + public void setBundleContext(final BundleContext bundleContext) { + this.bundleContext = bundleContext; + } - BrokerImpl.this.removeRpcImplementation(rpcType, implToRemove); - sessionRpcImpls.remove(rpcType); - } + /** + * @return the deactivator + */ + public AutoCloseable getDeactivator() { + return deactivator; + } - public Provider getProvider() { - return this.provider; - } + /** + * @param deactivator + * the deactivator to set + */ + public void setDeactivator(final AutoCloseable deactivator) { + this.deactivator = deactivator; + } + /** + * @return the router + */ + public RpcRouter getRouter() { + return router; } - public void setBundleContext(BundleContext context) { - this.bundleContext = context; + /** + * @param router + * the router to set + */ + public void setRouter(final RpcRouter router) { + this.router = router; } }