*/
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<ConsumerSessionImpl> sessions = Collections
- .synchronizedSet(new HashSet<ConsumerSessionImpl>());
- private Set<ProviderSessionImpl> providerSessions = Collections
- .synchronizedSet(new HashSet<ProviderSessionImpl>());
- private Set<BrokerModule> modules = Collections
- .synchronizedSet(new HashSet<BrokerModule>());
- private Map<Class<? extends BrokerService>, BrokerModule> serviceProviders = Collections
- .synchronizedMap(new HashMap<Class<? extends BrokerService>, BrokerModule>());
+ private final Set<ConsumerContextImpl> sessions = Collections
+ .synchronizedSet(new HashSet<ConsumerContextImpl>());
+ private final Set<ProviderContextImpl> providerSessions = Collections
+ .synchronizedSet(new HashSet<ProviderContextImpl>());
+
+ private AutoCloseable deactivator = null;
- // RPC Context
- private Map<QName, RpcImplementation> rpcImpls = Collections
- .synchronizedMap(new HashMap<QName, RpcImplementation>());
+ private RpcRouter router = null;
- // Implementation specific
- private ExecutorService executor;
+ private final ClassToInstanceMap<BrokerService> services;
+
+ public BrokerImpl(final RpcRouter router,final ClassToInstanceMap<BrokerService> 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<Class<? extends BrokerService>> provServices = module
- .getProvidedServices();
- for (Class<? extends BrokerService> serviceType : provServices) {
- log.info(" Registering session service implementation: "
- + serviceType.getCanonicalName());
- serviceProviders.put(serviceType, module);
- }
- }
-
- public <T extends BrokerService> T serviceFor(Class<T> 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<RpcResult<CompositeNode>> 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<RpcResult<CompositeNode>> invokeRpc(QName rpc,
- CompositeNode input) {
- RpcImplementation impl = rpcImpls.get(rpc);
- // if()
-
- Callable<RpcResult<CompositeNode>> call = callableFor(impl,
- rpc, input);
- Future<RpcResult<CompositeNode>> 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<RpcResult<CompositeNode>> callableFor(
- final RpcImplementation implemenation, final QName rpc,
- final CompositeNode input) {
-
- return new Callable<RpcResult<CompositeNode>>() {
-
- @Override
- public RpcResult<CompositeNode> call() throws Exception {
- return implemenation.invokeRpc(rpc, input);
- }
- };
+ @Override
+ public ListenerRegistration<RpcRegistrationListener> addRpcRegistrationListener(
+ final RpcRegistrationListener listener) {
+ return router.addRpcRegistrationListener(listener);
}
-
- private class ConsumerSessionImpl implements ConsumerSession {
-
- private final Consumer consumer;
-
- private Map<Class<? extends BrokerService>, BrokerService> instantiatedServices = Collections
- .synchronizedMap(new HashMap<Class<? extends BrokerService>, 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<RpcResult<CompositeNode>> rpc(QName rpc,
- CompositeNode input) {
- return BrokerImpl.this.invokeRpc(rpc, input);
- }
-
- @Override
- public <T extends BrokerService> T getService(Class<T> 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 void close() {
- Collection<BrokerService> toStop = instantiatedServices.values();
- this.closed = true;
- for (BrokerService brokerService : toStop) {
- brokerService.closeSession();
- }
- BrokerImpl.this.consumerSessionClosed(this);
- }
-
- @Override
- public boolean isClosed() {
- return closed;
- }
+ @Override
+ public <L extends RouteChangeListener<RpcRoutingContext, InstanceIdentifier>> ListenerRegistration<L> registerRouteChangeListener(
+ final L listener) {
+ return router.registerRouteChangeListener(listener);
}
- private class ProviderSessionImpl extends ConsumerSessionImpl implements
- ProviderSession {
-
- private Provider provider;
- private Map<QName, RpcImplementation> sessionRpcImpls = Collections.synchronizedMap(new HashMap<QName, RpcImplementation>());
-
- public ProviderSessionImpl(Provider provider, BundleContext ctx) {
- super(null,ctx);
- this.provider = provider;
- }
+ @Override
+ public Set<QName> getSupportedRpcs() {
+ return router.getSupportedRpcs();
+ }
- @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);
- }
+ @Override
+ public ListenableFuture<RpcResult<CompositeNode>> invokeRpc(
+ final QName rpc, final CompositeNode input) {
+ return router.invokeRpc(rpc, input);
+ }
- @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");
- }
+ /**
+ * @return the deactivator
+ */
+ public AutoCloseable getDeactivator() {
+ return deactivator;
+ }
- BrokerImpl.this.removeRpcImplementation(rpcType, implToRemove);
- sessionRpcImpls.remove(rpcType);
- }
+ /**
+ * @param deactivator
+ * the deactivator to set
+ */
+ public void setDeactivator(final AutoCloseable deactivator) {
+ this.deactivator = deactivator;
+ }
- public Provider getProvider() {
- return this.provider;
- }
+ /**
+ * @return the router
+ */
+ public RpcRouter getRouter() {
+ return router;
+ }
+ /**
+ * @param router
+ * the router to set
+ */
+ public void setRouter(final RpcRouter router) {
+ this.router = router;
}
- public void setBundleContext(BundleContext context) {
- this.bundleContext = context;
+ protected <T extends BrokerService> Optional<T> getGlobalService(final Class<T> service) {
+ return Optional.fromNullable(services.getInstance(service));
}
+
}
Code Review / controller.git / blobdiff