import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMultimap;
-import com.google.common.collect.ImmutableMultimap.Builder;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multimaps;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import javax.annotation.PreDestroy;
+import javax.inject.Inject;
+import org.eclipse.jdt.annotation.NonNull;
import org.opendaylight.mdsal.dom.api.DOMNotification;
import org.opendaylight.mdsal.dom.api.DOMNotificationListener;
import org.opendaylight.mdsal.dom.api.DOMNotificationPublishService;
import org.opendaylight.mdsal.dom.spi.DOMNotificationSubscriptionListener;
import org.opendaylight.mdsal.dom.spi.DOMNotificationSubscriptionListenerRegistry;
import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
+import org.opendaylight.yangtools.concepts.AbstractRegistration;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.concepts.Registration;
import org.opendaylight.yangtools.util.ListenerRegistry;
import org.opendaylight.yangtools.util.concurrent.EqualityQueuedNotificationManager;
import org.opendaylight.yangtools.util.concurrent.FluentFutures;
import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager;
import org.opendaylight.yangtools.yang.model.api.stmt.SchemaNodeIdentifier.Absolute;
+import org.osgi.service.component.annotations.Activate;
+import org.osgi.service.component.annotations.Component;
+import org.osgi.service.component.annotations.Deactivate;
+import org.osgi.service.metatype.annotations.AttributeDefinition;
+import org.osgi.service.metatype.annotations.Designate;
+import org.osgi.service.metatype.annotations.ObjectClassDefinition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
* Internal implementation one by using a {@link QueuedNotificationManager}.
*</p>
*/
+@Component(immediate = true, configurationPid = "org.opendaylight.mdsal.dom.notification", service = {
+ DOMNotificationService.class, DOMNotificationPublishService.class,
+ DOMNotificationSubscriptionListenerRegistry.class
+})
+@Designate(ocd = DOMNotificationRouter.Config.class)
+// Non-final for testing
public class DOMNotificationRouter implements AutoCloseable, DOMNotificationPublishService,
DOMNotificationService, DOMNotificationSubscriptionListenerRegistry {
+ @ObjectClassDefinition()
+ public @interface Config {
+ @AttributeDefinition(name = "notification-queue-depth")
+ int queueDepth() default 65536;
+ }
+
+ @VisibleForTesting
+ abstract static sealed class Reg<T extends DOMNotificationListener> extends AbstractListenerRegistration<T> {
+ Reg(final @NonNull T listener) {
+ super(listener);
+ }
+ }
+
+ private final class SingleReg<T extends DOMNotificationListener> extends Reg<T> {
+ SingleReg(final @NonNull T listener) {
+ super(listener);
+ }
+
+ @Override
+ protected void removeRegistration() {
+ DOMNotificationRouter.this.removeRegistration(this);
+ }
+ }
+
+ private static final class ComponentReg extends Reg<DOMNotificationListener> {
+ ComponentReg(final @NonNull DOMNotificationListener listener) {
+ super(listener);
+ }
+
+ @Override
+ protected void removeRegistration() {
+ // No-op
+ }
+ }
private static final Logger LOG = LoggerFactory.getLogger(DOMNotificationRouter.class);
private static final ListenableFuture<Void> NO_LISTENERS = FluentFutures.immediateNullFluentFuture();
private final ScheduledThreadPoolExecutor observer;
private final ExecutorService executor;
- private volatile Multimap<Absolute, AbstractListenerRegistration<? extends DOMNotificationListener>> listeners =
- ImmutableMultimap.of();
+ private volatile Multimap<Absolute, Reg<?>> listeners = ImmutableMultimap.of();
- @VisibleForTesting
- DOMNotificationRouter(int maxQueueCapacity) {
- observer = new ScheduledThreadPoolExecutor(1,
- new ThreadFactoryBuilder().setDaemon(true).setNameFormat("DOMNotificationRouter-observer-%d").build());
- executor = Executors.newCachedThreadPool(
- new ThreadFactoryBuilder().setDaemon(true).setNameFormat("DOMNotificationRouter-listeners-%d").build());
+ @Inject
+ public DOMNotificationRouter(final int maxQueueCapacity) {
+ observer = new ScheduledThreadPoolExecutor(1, new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat("DOMNotificationRouter-observer-%d")
+ .build());
+ executor = Executors.newCachedThreadPool(new ThreadFactoryBuilder()
+ .setDaemon(true)
+ .setNameFormat("DOMNotificationRouter-listeners-%d")
+ .build());
queueNotificationManager = new EqualityQueuedNotificationManager<>("DOMNotificationRouter", executor,
maxQueueCapacity, DOMNotificationRouter::deliverEvents);
}
- public static DOMNotificationRouter create(int maxQueueCapacity) {
+ @Activate
+ public DOMNotificationRouter(final Config config) {
+ this(config.queueDepth());
+ LOG.info("DOM Notification Router started");
+ }
+
+ @Deprecated(forRemoval = true)
+ public static DOMNotificationRouter create(final int maxQueueCapacity) {
return new DOMNotificationRouter(maxQueueCapacity);
}
@Override
public synchronized <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(
final T listener, final Collection<Absolute> types) {
- final AbstractListenerRegistration<T> reg = new AbstractListenerRegistration<>(listener) {
- @Override
- protected void removeRegistration() {
- synchronized (DOMNotificationRouter.this) {
- replaceListeners(ImmutableMultimap.copyOf(Multimaps.filterValues(listeners,
- input -> input != this)));
- }
- }
- };
+ final var reg = new SingleReg<>(listener);
if (!types.isEmpty()) {
- final Builder<Absolute, AbstractListenerRegistration<? extends DOMNotificationListener>> b =
- ImmutableMultimap.builder();
+ final var b = ImmutableMultimap.<Absolute, Reg<?>>builder();
b.putAll(listeners);
- for (final Absolute t : types) {
+ for (var t : types) {
b.put(t, reg);
}
return reg;
}
+ @Override
+ public synchronized Registration registerNotificationListeners(
+ final Map<Absolute, DOMNotificationListener> typeToListener) {
+ final var b = ImmutableMultimap.<Absolute, Reg<?>>builder();
+ b.putAll(listeners);
+
+ final var tmp = new HashMap<DOMNotificationListener, ComponentReg>();
+ for (var e : typeToListener.entrySet()) {
+ b.put(e.getKey(), tmp.computeIfAbsent(e.getValue(), ComponentReg::new));
+ }
+
+ final var regs = List.copyOf(tmp.values());
+ return new AbstractRegistration() {
+ @Override
+ protected void removeRegistration() {
+ regs.forEach(ComponentReg::close);
+ removeRegistrations(regs);
+ }
+ };
+ }
+
+ private synchronized void removeRegistration(final SingleReg<?> reg) {
+ replaceListeners(ImmutableMultimap.copyOf(Multimaps.filterValues(listeners, input -> input != reg)));
+ }
+
+ private synchronized void removeRegistrations(final List<ComponentReg> regs) {
+ replaceListeners(ImmutableMultimap.copyOf(Multimaps.filterValues(listeners, input -> !regs.contains(input))));
+ }
+
/**
* Swaps registered listeners and triggers notification update.
*
* @param newListeners is used to notify listenerTypes changed
*/
- private void replaceListeners(
- final Multimap<Absolute, AbstractListenerRegistration<? extends DOMNotificationListener>> newListeners) {
+ private void replaceListeners(final Multimap<Absolute, Reg<?>> newListeners) {
listeners = newListeners;
notifyListenerTypesChanged(newListeners.keySet());
}
return subscriptionListeners.register(listener);
}
-
@VisibleForTesting
- ListenableFuture<? extends Object> publish(DOMNotification notification,
- final Collection<AbstractListenerRegistration<? extends DOMNotificationListener>> subscribers) {
+ ListenableFuture<? extends Object> publish(final DOMNotification notification,
+ final Collection<Reg<?>> subscribers) {
final List<ListenableFuture<Void>> futures = new ArrayList<>(subscribers.size());
subscribers.forEach(subscriber -> {
final DOMNotificationRouterEvent event = new DOMNotificationRouterEvent(notification);
@Override
public ListenableFuture<? extends Object> putNotification(final DOMNotification notification)
throws InterruptedException {
- final Collection<AbstractListenerRegistration<? extends DOMNotificationListener>> subscribers =
- listeners.get(notification.getType());
+ final var subscribers = listeners.get(notification.getType());
if (subscribers.isEmpty()) {
return NO_LISTENERS;
}
@Override
public ListenableFuture<? extends Object> offerNotification(final DOMNotification notification) {
- final Collection<AbstractListenerRegistration<? extends DOMNotificationListener>> subscribers =
- listeners.get(notification.getType());
+ final var subscribers = listeners.get(notification.getType());
if (subscribers.isEmpty()) {
return NO_LISTENERS;
}
@Override
public ListenableFuture<? extends Object> offerNotification(final DOMNotification notification, final long timeout,
final TimeUnit unit) throws InterruptedException {
- final Collection<AbstractListenerRegistration<? extends DOMNotificationListener>> subscribers =
- listeners.get(notification.getType());
+ final var subscribers = listeners.get(notification.getType());
if (subscribers.isEmpty()) {
return NO_LISTENERS;
}
}
}
+ @PreDestroy
+ @Deactivate
@Override
public void close() {
observer.shutdown();
executor.shutdown();
+ LOG.info("DOM Notification Router stopped");
}
@VisibleForTesting
Code Review / mdsal.git / blobdiff