package org.opendaylight.groupbasedpolicy.endpoint;
+import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Executors;
import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
import org.opendaylight.controller.md.sal.binding.api.DataBroker;
import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
* Endpoint registry provides a scalable store for accessing and updating
* information about endpoints.
*/
-public class EndpointRpcRegistry implements EndpointService {
+public class EndpointRpcRegistry implements EndpointService, AutoCloseable {
private static final Logger LOG =
LoggerFactory.getLogger(EndpointRpcRegistry.class);
private final DataBroker dataProvider;
- private final ScheduledExecutorService executor;
private final RpcProviderRegistry rpcRegistry;
- private static EndpointRpcRegistry endpointRpcRegistry;
- final BindingAwareBroker.RpcRegistration<EndpointService> rpcRegistration;
+ private final BindingAwareBroker.RpcRegistration<EndpointService> rpcRegistration;
- private final static ConcurrentMap<String, EpRendererAugmentation> registeredRenderers = new ConcurrentHashMap<String, EpRendererAugmentation>();
+ final static ConcurrentMap<String, EpRendererAugmentation> registeredRenderers = new ConcurrentHashMap<String, EpRendererAugmentation>();
/**
* This method registers a renderer for endpoint RPC API. This method
* ensures single RPC registration for all renderers since a single RPC
* registration is only allowed.
*
- * @param dataProvider
- * - the dataProvider
- * @param rpcRegistry
- * - the rpcRegistry
- * @param executor
- * - thread pool executor
* @param epRendererAugmentation
* - specific implementation RPC augmentation, if any. Otherwise
* NULL
*/
- public static void register(DataBroker dataProvider,
- RpcProviderRegistry rpcRegistry,
- EpRendererAugmentation epRendererAugmentation) {
- if (dataProvider == null || rpcRegistry == null) {
- if (epRendererAugmentation != null) {
- LOG.warn("Couldn't register class {} for endpoint RPC because of missing required info", epRendererAugmentation);
- }
- return;
- }
- if (endpointRpcRegistry == null) {
- synchronized (EndpointRpcRegistry.class) {
- if (endpointRpcRegistry == null) {
- ScheduledExecutorService executor = Executors
- .newScheduledThreadPool(Math
- .max(Runtime.getRuntime()
- .availableProcessors() * 3, 10));
- endpointRpcRegistry = new EndpointRpcRegistry(dataProvider,
- rpcRegistry, executor);
- }
- }
- }
+ public static void register(EpRendererAugmentation epRendererAugmentation) {
if (epRendererAugmentation != null) {
- registeredRenderers.putIfAbsent(epRendererAugmentation.getClass()
- .getName(), epRendererAugmentation);
+ registeredRenderers.putIfAbsent(epRendererAugmentation.getClass().getName(), epRendererAugmentation);
+ LOG.info("Registered {}", epRendererAugmentation.getClass().getName());
}
}
/**
*
- * @param regImp
+ * @param regImp the endpoint augmentation
* @throws Exception
*/
- public static void unregister(EpRendererAugmentation regImp)
- throws Exception {
- if (regImp == null
- || !registeredRenderers
- .containsKey(regImp.getClass().getName())) {
+ public static void unregister(EpRendererAugmentation regImp) throws Exception {
+ if (regImp == null || !registeredRenderers.containsKey(regImp.getClass().getName())) {
return;
}
registeredRenderers.remove(regImp.getClass().getName());
LOG.info("Unregistered {}", regImp.getClass().getName());
- if (registeredRenderers.isEmpty() && endpointRpcRegistry != null) {
- synchronized (EndpointRpcRegistry.class) {
- if (registeredRenderers.isEmpty()
- && endpointRpcRegistry != null) {
- endpointRpcRegistry.rpcRegistration.close();
- endpointRpcRegistry.executor.shutdown();
- endpointRpcRegistry = null;
- }
- }
- }
- }
-
- public static Class<?> getAugmentationContextType(
- final Augmentation<?> augmentation) {
- if (augmentation == null) {
- return null;
- }
- final Class<?>[] augmentationInterfaces = augmentation.getClass()
- .getInterfaces();
- if (augmentationInterfaces.length == 1) {
- return augmentationInterfaces[0];
- }
- /*
- * if here, then the way YANG tools generate augmentation code has
- * changed, hence augmentation classes are not implemented by single
- * interface anymore. This is very unlikely to happen, but if it did, we
- * need to know about it in order to update this method.
- */
- LOG.error(
- "YANG Generated Code has Changed -- augmentation object {} is NOT implemented by one interface anymore",
- augmentation);
- return null;
}
/**
* Constructor
*
- * @param dataProvider
- * @param rpcRegistry
- * @param executor
+ * @param dataProvider the {@link DataBroker}
+ * @param rpcRegistry the {@link RpcProviderRegistry}
*/
- private EndpointRpcRegistry(DataBroker dataProvider,
- RpcProviderRegistry rpcRegistry,
- ScheduledExecutorService executor) {
+ public EndpointRpcRegistry(DataBroker dataProvider, RpcProviderRegistry rpcRegistry) {
this.dataProvider = dataProvider;
- this.executor = executor;
this.rpcRegistry = rpcRegistry;
-
if (this.rpcRegistry != null) {
rpcRegistration =
this.rpcRegistry.addRpcImplementation(EndpointService.class, this);
// endpoint group/condition mappings with no conditions
}
+ @Override
+ public void close() throws Exception {
+ if (rpcRegistration != null) {
+ rpcRegistration.close();
+ }
+ }
+
/**
* Construct an endpoint with the appropriate augmentations from the
* endpoint input. Each concrete implementation can provides its specifics
for (Entry<String, EpRendererAugmentation> entry : registeredRenderers
.entrySet()) {
try {
- Augmentation<Endpoint> augmentation = entry.getValue()
- .buildEndpointAugmentation(input);
- if (augmentation != null) {
- @SuppressWarnings("unchecked")
- Class<? extends Augmentation<Endpoint>> augmentationType = (Class<? extends Augmentation<Endpoint>>) getAugmentationContextType(augmentation);
- eb.addAugmentation(augmentationType, augmentation);
+ Map.Entry<Class<? extends Augmentation<Endpoint>>, Augmentation<Endpoint>> augmentationEntry =
+ entry.getValue().buildEndpointAugmentation(input);
+ if (augmentationEntry != null) {
+ eb.addAugmentation(augmentationEntry.getKey(), augmentationEntry.getValue());
}
- } catch (Throwable t) {
+ } catch (Exception e) {
LOG.warn("Endpoint Augmentation error while processing "
- + entry.getKey() + ". Reason: ", t);
+ + entry.getKey() + ". Reason: ", e);
}
}
return eb;
for (Entry<String, EpRendererAugmentation> entry : registeredRenderers
.entrySet()) {
try {
- Augmentation<EndpointL3> augmentation = entry.getValue()
- .buildEndpointL3Augmentation(input);
- if (augmentation != null) {
- @SuppressWarnings("unchecked")
- Class<? extends Augmentation<EndpointL3>> augmentationType = (Class<? extends Augmentation<EndpointL3>>) getAugmentationContextType(augmentation);
- eb.addAugmentation(augmentationType, augmentation);
+ Map.Entry<Class<? extends Augmentation<EndpointL3>>, Augmentation<EndpointL3>> augmentationEntry =
+ entry.getValue().buildEndpointL3Augmentation(input);
+ if (augmentationEntry != null) {
+ eb.addAugmentation(augmentationEntry.getKey(), augmentationEntry.getValue());
}
- } catch (Throwable t) {
+ } catch (Exception e) {
LOG.warn("L3 endpoint Augmentation error while processing "
- + entry.getKey() + ". Reason: ", t);
+ + entry.getKey() + ". Reason: ", e);
}
}
return eb;
for (Entry<String, EpRendererAugmentation> entry : registeredRenderers
.entrySet()) {
try {
- entry.getValue().buildL3PrefixEndpointAugmentation(eb, input);
- } catch (Throwable t) {
+ Map.Entry<Class<? extends Augmentation<EndpointL3Prefix>>, Augmentation<EndpointL3Prefix>> augmentationEntry =
+ entry.getValue().buildL3PrefixEndpointAugmentation(input);
+ if (augmentationEntry != null) {
+ eb.addAugmentation(augmentationEntry.getKey(), augmentationEntry.getValue());
+ }
+ } catch (Exception e) {
LOG.warn("L3 endpoint Augmentation error while processing "
- + entry.getKey() + ". Reason: ", t);
+ + entry.getKey() + ". Reason: ", e);
}
}
return eb;
InstanceIdentifier.builder(Endpoints.class)
.child(Endpoint.class, key)
.build();
- t.put(LogicalDatastoreType.OPERATIONAL, iid, ep);
+ t.put(LogicalDatastoreType.OPERATIONAL, iid, ep, true);
}
if (input.getL3Address() != null) {
for (L3Address l3addr : input.getL3Address()) {
InstanceIdentifier.builder(Endpoints.class)
.child(EndpointL3.class, key3)
.build();
- t.put(LogicalDatastoreType.OPERATIONAL, iid_l3, ep3);
+ t.put(LogicalDatastoreType.OPERATIONAL, iid_l3, ep3, true);
}
}
ListenableFuture<Void> r = t.submit();
- return Futures.transform(r, futureTrans, executor);
+ return Futures.transform(r, futureTrans);
}
@Override
t.put(LogicalDatastoreType.OPERATIONAL, iid_l3prefix, epL3Prefix);
ListenableFuture<Void> r = t.submit();
- return Futures.transform(r, futureTrans, executor);
+ return Futures.transform(r, futureTrans);
}
@Override
// TODO: Implement L3Prefix
ListenableFuture<Void> r = t.submit();
- return Futures.transform(r, futureTrans, executor);
+ return Futures.transform(r, futureTrans);
}
@Override
}
ListenableFuture<Void> r = t.submit();
- return Futures.transform(r, futureTrans, executor);
+ return Futures.transform(r, futureTrans);
}
@Override
}
ListenableFuture<Void> r = t.submit();
- return Futures.transform(r, futureTrans, executor);
+ return Futures.transform(r, futureTrans);
}
Function<Void, RpcResult<Void>> futureTrans =
return RpcResultBuilder.<Void> success().build();
}
};
-
}
Code Review / groupbasedpolicy.git / blobdiff