--- /dev/null
+/*
+ * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+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.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.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;
+
+public class BrokerImpl implements Broker {
+ private 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>());
+
+ // RPC Context
+ private Map<QName, RpcImplementation> rpcImpls = Collections
+ .synchronizedMap(new HashMap<QName, RpcImplementation>());
+
+ // Implementation specific
+ private ExecutorService executor;
+
+ private BundleContext bundleContext;
+
+ @Override
+ public ConsumerSession registerConsumer(Consumer consumer,BundleContext ctx) {
+ checkPredicates(consumer);
+ log.info("Registering consumer " + consumer);
+ ConsumerSessionImpl session = newSessionFor(consumer,ctx);
+ consumer.onSessionInitiated(session);
+ sessions.add(session);
+ return session;
+ }
+
+ @Override
+ public ProviderSession registerProvider(Provider provider,BundleContext ctx) {
+ checkPredicates(provider);
+
+ ProviderSessionImpl 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);
+ }
+
+ // 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
+ }
+
+ 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;
+ }
+
+ // 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 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");
+ }
+ }
+
+ // Private Factory methods
+
+ private ConsumerSessionImpl newSessionFor(Consumer provider, BundleContext ctx) {
+ return new ConsumerSessionImpl(provider,ctx);
+ }
+
+ private ProviderSessionImpl newSessionFor(Provider provider, BundleContext ctx) {
+ return new ProviderSessionImpl(provider,ctx);
+ }
+
+ private void consumerSessionClosed(ConsumerSessionImpl consumerSessionImpl) {
+ sessions.remove(consumerSessionImpl);
+ providerSessions.remove(consumerSessionImpl);
+ }
+
+ 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);
+ }
+ };
+ }
+
+ 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;
+ }
+
+ }
+
+ 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 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 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);
+ }
+
+ public Provider getProvider() {
+ return this.provider;
+ }
+
+ }
+
+ public void setBundleContext(BundleContext context) {
+ this.bundleContext = context;
+ }
+}
Code Review / controller.git / blobdiff