*/
package org.opendaylight.mdsal.dom.broker;
+import static java.util.Objects.requireNonNull;
+
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMultimap;
+import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Multimaps;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
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.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.annotation.PreDestroy;
import javax.inject.Inject;
+import javax.inject.Singleton;
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.DOMNotificationPublishDemandExtension;
+import org.opendaylight.mdsal.dom.api.DOMNotificationPublishDemandExtension.DemandListener;
import org.opendaylight.mdsal.dom.api.DOMNotificationPublishService;
import org.opendaylight.mdsal.dom.api.DOMNotificationService;
-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.ObjectRegistry;
import org.opendaylight.yangtools.util.concurrent.EqualityQueuedNotificationManager;
import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager;
import org.opendaylight.yangtools.yang.common.Empty;
* Internal implementation one by using a {@link QueuedNotificationManager}.
*</p>
*/
-@Component(configurationPid = "org.opendaylight.mdsal.dom.notification", service = {
- DOMNotificationRouter.class, DOMNotificationSubscriptionListenerRegistry.class
-})
+@Singleton
+@Component(configurationPid = "org.opendaylight.mdsal.dom.notification", service = DOMNotificationRouter.class)
@Designate(ocd = DOMNotificationRouter.Config.class)
// Non-final for testing
-public class DOMNotificationRouter implements DOMNotificationSubscriptionListenerRegistry, AutoCloseable {
+public class DOMNotificationRouter implements AutoCloseable {
@ObjectClassDefinition()
public @interface Config {
@AttributeDefinition(name = "notification-queue-depth")
}
@VisibleForTesting
- abstract static sealed class Reg<T extends DOMNotificationListener> extends AbstractListenerRegistration<T> {
- Reg(final @NonNull T listener) {
- super(listener);
+ abstract static sealed class Reg extends AbstractRegistration {
+ private final @NonNull DOMNotificationListener listener;
+
+ Reg(final @NonNull DOMNotificationListener listener) {
+ this.listener = requireNonNull(listener);
}
}
- private final class SingleReg<T extends DOMNotificationListener> extends Reg<T> {
- SingleReg(final @NonNull T listener) {
+ private final class SingleReg extends Reg {
+ SingleReg(final @NonNull DOMNotificationListener listener) {
super(listener);
}
}
}
- private static final class ComponentReg extends Reg<DOMNotificationListener> {
+ private static final class ComponentReg extends Reg {
ComponentReg(final @NonNull DOMNotificationListener listener) {
super(listener);
}
}
}
- private final class PublishFacade implements DOMNotificationPublishService {
+ private final class PublishFacade implements DOMNotificationPublishService, DOMNotificationPublishDemandExtension {
+ @Override
+ public List<Extension> supportedExtensions() {
+ return List.of(this);
+ }
+
@Override
public ListenableFuture<? extends Object> putNotification(final DOMNotification notification)
throws InterruptedException {
return DOMNotificationPublishService.REJECTED;
}
}
+
+ @Override
+ public Registration registerDemandListener(final DemandListener listener) {
+ final var initialTypes = listeners.keySet();
+ executor.execute(() -> listener.onDemandUpdated(initialTypes));
+ return demandListeners.register(listener);
+ }
}
private final class SubscribeFacade implements DOMNotificationService {
@Override
- public <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(
- final
T listener, final Collection<Absolute> types) {
+ public Registration registerNotificationListener(final DOMNotificationListener listener,
+ final Collection<Absolute> types) {
synchronized (DOMNotificationRouter.this) {
- final var reg = new SingleReg<>(listener);
+ final var reg = new SingleReg(listener);
if (!types.isEmpty()) {
- final var b = ImmutableMultimap.<Absolute, Reg<?>>builder();
+ final var b = ImmutableMultimap.<Absolute, Reg>builder();
b.putAll(listeners);
for (var t : types) {
public synchronized Registration registerNotificationListeners(
final Map<Absolute, DOMNotificationListener> typeToListener) {
synchronized (DOMNotificationRouter.this) {
- final var b = ImmutableMultimap.<Absolute, Reg<?>>builder();
+ final var b = ImmutableMultimap.<Absolute, Reg>builder();
b.putAll(listeners);
final var tmp = new HashMap<DOMNotificationListener, ComponentReg>();
private static final Logger LOG = LoggerFactory.getLogger(DOMNotificationRouter.class);
private static final @NonNull ListenableFuture<?> NO_LISTENERS = Futures.immediateFuture(Empty.value());
- private final ListenerRegistry<DOMNotificationSubscriptionListener> subscriptionListeners =
- ListenerRegistry.create();
- private final EqualityQueuedNotificationManager<AbstractListenerRegistration<? extends DOMNotificationListener>,
- DOMNotificationRouterEvent> queueNotificationManager;
+ private final EqualityQueuedNotificationManager<Reg, DOMNotificationRouterEvent> queueNotificationManager;
private final @NonNull DOMNotificationPublishService notificationPublishService = new PublishFacade();
private final @NonNull DOMNotificationService notificationService = new SubscribeFacade();
+ private final ObjectRegistry<DemandListener> demandListeners =
+ ObjectRegistry.createConcurrent("notification demand listeners");
private final ScheduledThreadPoolExecutor observer;
private final ExecutorService executor;
- private volatile ImmutableMultimap<Absolute, Reg<?>> listeners = ImmutableMultimap.of();
+ private volatile ImmutableMultimap<Absolute, Reg> listeners = ImmutableMultimap.of();
@Inject
public DOMNotificationRouter(final int maxQueueCapacity) {
return notificationPublishService;
}
- private synchronized void removeRegistration(final SingleReg<?> reg) {
+ private synchronized void removeRegistration(final SingleReg reg) {
replaceListeners(ImmutableMultimap.copyOf(Multimaps.filterValues(listeners, input -> input != reg)));
}
*
* @param newListeners is used to notify listenerTypes changed
*/
- private void replaceListeners(final ImmutableMultimap<Absolute, Reg<?>> newListeners) {
+ private void replaceListeners(final ImmutableMultimap<Absolute, Reg> newListeners) {
listeners = newListeners;
notifyListenerTypesChanged(newListeners.keySet());
}
@SuppressWarnings("checkstyle:IllegalCatch")
- private void notifyListenerTypesChanged(final Set<Absolute> typesAfter) {
- final var listenersAfter = subscriptionListeners.streamListeners().collect(ImmutableList.toImmutableList());
+ private void notifyListenerTypesChanged(final
@NonNull ImmutableSet<Absolute> typesAfter) {
+ final var listenersAfter = demandListeners.streamObjects().collect(ImmutableList.toImmutableList());
executor.execute(() -> {
- for (var subListener : listenersAfter) {
+ for (var listener : listenersAfter) {
try {
- subListener.onSubscriptionChanged(typesAfter);
+ listener.onDemandUpdated(typesAfter);
} catch (final Exception e) {
- LOG.warn("Uncaught exception during invoking listener {}", subListener, e);
+ LOG.warn("Uncaught exception during invoking listener {}", listener, e);
}
}
});
}
- @Override
- public <L extends DOMNotificationSubscriptionListener> ListenerRegistration<L> registerSubscriptionListener(
- final L listener) {
- final var initialTypes = listeners.keySet();
- executor.execute(() -> listener.onSubscriptionChanged(initialTypes));
- return subscriptionListeners.register(listener);
- }
-
@VisibleForTesting
@NonNull ListenableFuture<? extends Object> putNotificationImpl(final DOMNotification notification)
throws InterruptedException {
}
@VisibleForTesting
- @NonNull ListenableFuture<?> publish(final DOMNotification notification, final Collection<Reg<?>> subscribers) {
+ @NonNull ListenableFuture<?> publish(final DOMNotification notification, final Collection<Reg> subscribers) {
final var futures = new ArrayList<ListenableFuture<?>>(subscribers.size());
subscribers.forEach(subscriber -> {
final var event = new DOMNotificationRouterEvent(notification);
}
@VisibleForTesting
- ListenerRegistry<DOMNotificationSubscriptionListener> subscriptionListeners() {
- return subscriptionListeners;
+ ObjectRegistry<DemandListener> demandListeners() {
+ return demandListeners;
}
- private static void deliverEvents(final AbstractListenerRegistration<? extends DOMNotificationListener> reg,
- final ImmutableList<DOMNotificationRouterEvent> events) {
+ private static void deliverEvents(final Reg reg, final ImmutableList<DOMNotificationRouterEvent> events) {
if (reg.notClosed()) {
- final var listener = reg.getInstance();
+ final var listener = reg.listener;
for (var event : events) {
event.deliverTo(listener);
}
Code Review / mdsal.git / blobdiff