Proxy DOMNotificationRouter to the mdsal implementation 81/73381/2
authorTom Pantelis <tompantelis@gmail.com>
Sat, 23 Jun 2018 19:33:37 +0000 (15:33 -0400)
committerTom Pantelis <tompantelis@gmail.com>
Mon, 25 Jun 2018 13:32:09 +0000 (09:32 -0400)
Proxying to the dom mdsal notification service(s) will allow
an easier migration path where both can co-exist.

Change-Id: I514c319d457bce3261e6f9b36cb189828800fe97
Signed-off-by: Tom Pantelis <tompantelis@gmail.com>
opendaylight/md-sal/sal-dom-api/src/main/java/org/opendaylight/controller/md/sal/dom/api/DOMEvent.java
opendaylight/md-sal/sal-dom-api/src/main/java/org/opendaylight/controller/md/sal/dom/api/DOMNotification.java
opendaylight/md-sal/sal-dom-broker/src/main/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouter.java
opendaylight/md-sal/sal-dom-broker/src/main/resources/org/opendaylight/blueprint/dom-broker.xml
opendaylight/md-sal/sal-dom-broker/src/test/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouterTest.java [new file with mode: 0644]
opendaylight/md-sal/sal-dom-spi/src/main/java/org/opendaylight/controller/md/sal/dom/spi/DOMNotificationSubscriptionListener.java

index 6022b184dc285fde4a7119f8f8b0680d5f7cf698..312de34d611849fe1056dbcae77c495970899a31 100644 (file)
@@ -8,12 +8,21 @@
 
 package org.opendaylight.controller.md.sal.dom.api;
 
 
 package org.opendaylight.controller.md.sal.dom.api;
 
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.time.Instant;
 import java.util.Date;
 
 /**
  * Generic event interface.
  */
 import java.util.Date;
 
 /**
  * Generic event interface.
  */
-public interface DOMEvent {
+@SuppressFBWarnings(value = "NM_SAME_SIMPLE_NAME_AS_INTERFACE", justification = "Migration")
+public interface DOMEvent extends org.opendaylight.mdsal.dom.api.DOMEvent {
+
+    @Override
+    default Instant getEventInstant() {
+        final Date eventTime = getEventTime();
+        return eventTime != null ? eventTime.toInstant() : null;
+    }
 
     /**
      * Get the time of the event occurrence.
 
     /**
      * Get the time of the event occurrence.
index d99001ea5c9b44be38d6237f39ad5f507df955e7..d30e0f243fa137d82e148a86b1eed8b6665f8c20 100644 (file)
@@ -7,25 +7,11 @@
  */
 package org.opendaylight.controller.md.sal.dom.api;
 
  */
 package org.opendaylight.controller.md.sal.dom.api;
 
-import javax.annotation.Nonnull;
-import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
-import org.opendaylight.yangtools.yang.model.api.SchemaPath;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 
 /**
  * A single YANG notification.
  */
 
 /**
  * A single YANG notification.
  */
-public interface DOMNotification {
-    /**
-     * Return the type of this notification.
-     *
-     * @return Notification type.
-     */
-    @Nonnull SchemaPath getType();
-
-    /**
-     * Return the body of this notification.
-     *
-     * @return Notification body.
-     */
-    @Nonnull ContainerNode getBody();
+@SuppressFBWarnings(value = "NM_SAME_SIMPLE_NAME_AS_INTERFACE", justification = "Migration")
+public interface DOMNotification extends org.opendaylight.mdsal.dom.api.DOMNotification {
 }
 }
index 4e12f93d2d42b83b8759256fdfb61ca3cbc73db3..b5371fc289dd0a4cacf0195becb70d849557d13c 100644 (file)
@@ -7,28 +7,14 @@
  */
 package org.opendaylight.controller.md.sal.dom.broker.impl;
 
  */
 package org.opendaylight.controller.md.sal.dom.broker.impl;
 
-import com.google.common.base.Preconditions;
-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.ListenableFuture;
 import com.google.common.util.concurrent.ListenableFuture;
-import com.lmax.disruptor.EventHandler;
-import com.lmax.disruptor.InsufficientCapacityException;
-import com.lmax.disruptor.PhasedBackoffWaitStrategy;
-import com.lmax.disruptor.WaitStrategy;
-import com.lmax.disruptor.dsl.Disruptor;
-import com.lmax.disruptor.dsl.ProducerType;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.time.Instant;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.List;
-import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
+import java.util.Date;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeUnit;
+import org.opendaylight.controller.md.sal.dom.api.DOMEvent;
 import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationPublishService;
 import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
 import org.opendaylight.controller.md.sal.dom.api.DOMNotificationPublishService;
@@ -37,10 +23,8 @@ import org.opendaylight.controller.md.sal.dom.spi.DOMNotificationSubscriptionLis
 import org.opendaylight.controller.md.sal.dom.spi.DOMNotificationSubscriptionListenerRegistry;
 import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
 import org.opendaylight.yangtools.concepts.ListenerRegistration;
 import org.opendaylight.controller.md.sal.dom.spi.DOMNotificationSubscriptionListenerRegistry;
 import org.opendaylight.yangtools.concepts.AbstractListenerRegistration;
 import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.util.ListenerRegistry;
+import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
 import org.opendaylight.yangtools.yang.model.api.SchemaPath;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Joint implementation of {@link DOMNotificationPublishService} and {@link DOMNotificationService}. Provides
 
 /**
  * Joint implementation of {@link DOMNotificationPublishService} and {@link DOMNotificationService}. Provides
@@ -54,9 +38,8 @@ import org.slf4j.LoggerFactory;
  * on this instance, notifications do not take any locks here.
  *
  * <p>
  * on this instance, notifications do not take any locks here.
  *
  * <p>
- * The fully-blocking {@link #publish(long, DOMNotification, Collection)} and non-blocking
- * {@link #offerNotification(DOMNotification)}
- * are realized using the Disruptor's native operations. The bounded-blocking
+ * The fully-blocking {@link #offerNotification(DOMNotification)}
+ * is realized using the Disruptor's native operations. The bounded-blocking
  * {@link #offerNotification(DOMNotification, long, TimeUnit)}
  * is realized by arming a background wakeup interrupt.
  */
  * {@link #offerNotification(DOMNotification, long, TimeUnit)}
  * is realized by arming a background wakeup interrupt.
  */
@@ -64,190 +47,134 @@ import org.slf4j.LoggerFactory;
 public final class DOMNotificationRouter implements AutoCloseable, DOMNotificationPublishService,
         DOMNotificationService, DOMNotificationSubscriptionListenerRegistry {
 
 public final class DOMNotificationRouter implements AutoCloseable, DOMNotificationPublishService,
         DOMNotificationService, DOMNotificationSubscriptionListenerRegistry {
 
-    private static final Logger LOG = LoggerFactory.getLogger(DOMNotificationRouter.class);
-    private static final ListenableFuture<Void> NO_LISTENERS = Futures.immediateFuture(null);
-    private static final WaitStrategy DEFAULT_STRATEGY = PhasedBackoffWaitStrategy
-            .withLock(1L, 30L, TimeUnit.MILLISECONDS);
-    private static final EventHandler<DOMNotificationRouterEvent> DISPATCH_NOTIFICATIONS
-            = (event, sequence, endOfBatch) -> event.deliverNotification();
-    private static final EventHandler<DOMNotificationRouterEvent> NOTIFY_FUTURE = (event, sequence, endOfBatch) -> event
-            .setFuture();
-
-    private final Disruptor<DOMNotificationRouterEvent> disruptor;
-    private final ExecutorService executor;
-    private volatile Multimap<SchemaPath, ListenerRegistration<? extends DOMNotificationListener>> listeners
-            = ImmutableMultimap.of();
-    private final ListenerRegistry<DOMNotificationSubscriptionListener> subscriptionListeners = ListenerRegistry
-            .create();
-
-    @SuppressWarnings("unchecked")
-    private DOMNotificationRouter(final ExecutorService executor, final int queueDepth, final WaitStrategy strategy) {
-        this.executor = Preconditions.checkNotNull(executor);
-
-        disruptor = new Disruptor<>(DOMNotificationRouterEvent.FACTORY, queueDepth, executor, ProducerType.MULTI,
-                                    strategy);
-        disruptor.handleEventsWith(DISPATCH_NOTIFICATIONS);
-        disruptor.after(DISPATCH_NOTIFICATIONS).handleEventsWith(NOTIFY_FUTURE);
-        disruptor.start();
+    private final org.opendaylight.mdsal.dom.api.DOMNotificationService delegateNotificationService;
+    private final org.opendaylight.mdsal.dom.api.DOMNotificationPublishService delegateNotificationPublishService;
+    private final org.opendaylight.mdsal.dom.spi.DOMNotificationSubscriptionListenerRegistry delegateListenerRegistry;
+
+    private DOMNotificationRouter(
+            org.opendaylight.mdsal.dom.api.DOMNotificationService delegateNotificationService,
+            org.opendaylight.mdsal.dom.api.DOMNotificationPublishService delegateNotificationPublishService,
+            org.opendaylight.mdsal.dom.spi.DOMNotificationSubscriptionListenerRegistry delegateListenerRegistry) {
+        this.delegateNotificationService = delegateNotificationService;
+        this.delegateNotificationPublishService = delegateNotificationPublishService;
+        this.delegateListenerRegistry = delegateListenerRegistry;
     }
 
     public static DOMNotificationRouter create(final int queueDepth) {
     }
 
     public static DOMNotificationRouter create(final int queueDepth) {
-        final ExecutorService executor = Executors.newCachedThreadPool();
-
-        return new DOMNotificationRouter(executor, queueDepth, DEFAULT_STRATEGY);
+        final org.opendaylight.mdsal.dom.broker.DOMNotificationRouter delegate =
+                org.opendaylight.mdsal.dom.broker.DOMNotificationRouter.create(queueDepth);
+        return create(delegate, delegate, delegate);
     }
 
     public static DOMNotificationRouter create(final int queueDepth, final long spinTime, final long parkTime,
                                                final TimeUnit unit) {
     }
 
     public static DOMNotificationRouter create(final int queueDepth, final long spinTime, final long parkTime,
                                                final TimeUnit unit) {
-        Preconditions.checkArgument(Long.lowestOneBit(queueDepth) == Long.highestOneBit(queueDepth),
-                                    "Queue depth %s is not power-of-two", queueDepth);
-        final ExecutorService executor = Executors.newCachedThreadPool();
-        final WaitStrategy strategy = PhasedBackoffWaitStrategy.withLock(spinTime, parkTime, unit);
+        final org.opendaylight.mdsal.dom.broker.DOMNotificationRouter delegate =
+                org.opendaylight.mdsal.dom.broker.DOMNotificationRouter.create(queueDepth, spinTime, parkTime, unit);
+        return create(delegate, delegate, delegate);
+    }
 
 
-        return new DOMNotificationRouter(executor, queueDepth, strategy);
+    public static DOMNotificationRouter create(
+            final org.opendaylight.mdsal.dom.api.DOMNotificationService delegateNotificationService,
+            final org.opendaylight.mdsal.dom.api.DOMNotificationPublishService delegateNotificationPublishService,
+            final org.opendaylight.mdsal.dom.spi.DOMNotificationSubscriptionListenerRegistry delegateListenerRegistry) {
+        return new DOMNotificationRouter(delegateNotificationService, delegateNotificationPublishService,
+                delegateListenerRegistry);
     }
 
     @Override
     public synchronized <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(
             final T listener, final Collection<SchemaPath> types) {
     }
 
     @Override
     public synchronized <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(
             final T listener, final Collection<SchemaPath> types) {
-        final ListenerRegistration<T> reg = new AbstractListenerRegistration<T>(listener) {
-            @Override
-            protected void removeRegistration() {
-                final ListenerRegistration<T> me = this;
+        org.opendaylight.mdsal.dom.api.DOMNotificationListener delegateListener = notification -> {
+            if (notification instanceof DOMNotification) {
+                listener.onNotification((DOMNotification)notification);
+                return;
+            }
 
 
-                synchronized (DOMNotificationRouter.this) {
-                    replaceListeners(ImmutableMultimap.copyOf(Multimaps.filterValues(listeners, input -> input != me)));
-                }
+            if (notification instanceof org.opendaylight.mdsal.dom.api.DOMEvent) {
+                listener.onNotification(new DefaultDOMEvent(notification,
+                        (org.opendaylight.mdsal.dom.api.DOMEvent)notification));
+                return;
             }
             }
+
+            listener.onNotification(new DefaultDOMNotification(notification));
         };
 
         };
 
-        if (!types.isEmpty()) {
-            final Builder<SchemaPath, ListenerRegistration<? extends DOMNotificationListener>> b = ImmutableMultimap
-                    .builder();
-            b.putAll(listeners);
+        final ListenerRegistration<org.opendaylight.mdsal.dom.api.DOMNotificationListener> reg =
+                delegateNotificationService.registerNotificationListener(delegateListener, types);
 
 
-            for (final SchemaPath t : types) {
-                b.put(t, reg);
+        return new AbstractListenerRegistration<T>(listener) {
+            @Override
+            protected void removeRegistration() {
+                reg.close();
             }
             }
-
-            replaceListeners(b.build());
-        }
-
-        return reg;
+        };
     }
 
     @Override
     public <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(final T listener,
     }
 
     @Override
     public <T extends DOMNotificationListener> ListenerRegistration<T> registerNotificationListener(final T listener,
-                                                                                                    final
-                                                                                                    SchemaPath...
-                                                                                                            types) {
+            final SchemaPath... types) {
         return registerNotificationListener(listener, Arrays.asList(types));
     }
 
         return registerNotificationListener(listener, Arrays.asList(types));
     }
 
-    /**
-     * Swaps registered listeners and triggers notification update.
-     *
-     * @param newListeners listeners
-     */
-    private void replaceListeners(
-            final Multimap<SchemaPath, ListenerRegistration<? extends DOMNotificationListener>> newListeners) {
-        listeners = newListeners;
-        notifyListenerTypesChanged(newListeners.keySet());
-    }
-
-    @SuppressWarnings("checkstyle:IllegalCatch")
-    private void notifyListenerTypesChanged(final Set<SchemaPath> typesAfter) {
-        final List<ListenerRegistration<DOMNotificationSubscriptionListener>> listenersAfter = ImmutableList
-                .copyOf(subscriptionListeners.getListeners());
-        executor.execute(() -> {
-            for (final ListenerRegistration<DOMNotificationSubscriptionListener> subListener : listenersAfter) {
-                try {
-                    subListener.getInstance().onSubscriptionChanged(typesAfter);
-                } catch (final Exception e) {
-                    LOG.warn("Uncaught exception during invoking listener {}", subListener.getInstance(), e);
-                }
-            }
-        });
-    }
-
     @Override
     public <L extends DOMNotificationSubscriptionListener> ListenerRegistration<L> registerSubscriptionListener(
             final L listener) {
     @Override
     public <L extends DOMNotificationSubscriptionListener> ListenerRegistration<L> registerSubscriptionListener(
             final L listener) {
-        final Set<SchemaPath> initialTypes = listeners.keySet();
-        executor.execute(() -> listener.onSubscriptionChanged(initialTypes));
-        return subscriptionListeners.registerWithType(listener);
-    }
-
-    private ListenableFuture<Void> publish(final long seq, final DOMNotification notification,
-                                           final Collection<ListenerRegistration<? extends DOMNotificationListener>>
-                                                   subscribers) {
-        final DOMNotificationRouterEvent event = disruptor.get(seq);
-        final ListenableFuture<Void> future = event.initialize(notification, subscribers);
-        disruptor.getRingBuffer().publish(seq);
-        return future;
+        return delegateListenerRegistry.registerSubscriptionListener(listener);
     }
 
     @Override
     public ListenableFuture<?> putNotification(final DOMNotification notification) throws InterruptedException {
     }
 
     @Override
     public ListenableFuture<?> putNotification(final DOMNotification notification) throws InterruptedException {
-        final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers = listeners
-                .get(notification.getType());
-        if (subscribers.isEmpty()) {
-            return NO_LISTENERS;
-        }
-
-        final long seq = disruptor.getRingBuffer().next();
-        return publish(seq, notification, subscribers);
-    }
-
-    private ListenableFuture<?> tryPublish(final DOMNotification notification,
-                                           final Collection<ListenerRegistration<? extends DOMNotificationListener>>
-                                                   subscribers) {
-        final long seq;
-        try {
-            seq = disruptor.getRingBuffer().tryNext();
-        } catch (final InsufficientCapacityException e) {
-            return DOMNotificationPublishService.REJECTED;
-        }
-
-        return publish(seq, notification, subscribers);
+        return delegateNotificationPublishService.putNotification(notification);
     }
 
     @Override
     public ListenableFuture<?> offerNotification(final DOMNotification notification) {
     }
 
     @Override
     public ListenableFuture<?> offerNotification(final DOMNotification notification) {
-        final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers = listeners
-                .get(notification.getType());
-        if (subscribers.isEmpty()) {
-            return NO_LISTENERS;
-        }
-
-        return tryPublish(notification, subscribers);
+        return delegateNotificationPublishService.offerNotification(notification);
     }
 
     @Override
     public ListenableFuture<?> offerNotification(final DOMNotification notification, final long timeout,
                                                  final TimeUnit unit) throws InterruptedException {
     }
 
     @Override
     public ListenableFuture<?> offerNotification(final DOMNotification notification, final long timeout,
                                                  final TimeUnit unit) throws InterruptedException {
-        final Collection<ListenerRegistration<? extends DOMNotificationListener>> subscribers = listeners
-                .get(notification.getType());
-        if (subscribers.isEmpty()) {
-            return NO_LISTENERS;
+        return delegateNotificationPublishService.offerNotification(notification, timeout, unit);
+    }
+
+    @Override
+    public void close() {
+    }
+
+    private static class DefaultDOMNotification implements DOMNotification {
+        private final SchemaPath schemaPath;
+        private final ContainerNode body;
+
+        DefaultDOMNotification(org.opendaylight.mdsal.dom.api.DOMNotification from) {
+            this.schemaPath = from.getType();
+            this.body = from.getBody();
         }
 
         }
 
-        // Attempt to perform a non-blocking publish first
-        final ListenableFuture<?> noBlock = tryPublish(notification, subscribers);
-        if (!DOMNotificationPublishService.REJECTED.equals(noBlock)) {
-            return noBlock;
+        @Override
+        public SchemaPath getType() {
+            return schemaPath;
         }
 
         }
 
-        /*
-         * FIXME: we need a background thread, which will watch out for blocking too long. Here
-         *        we will arm a tasklet for it and synchronize delivery of interrupt properly.
-         */
-        throw new UnsupportedOperationException("Not implemented yet");
+        @Override
+        public ContainerNode getBody() {
+            return body;
+        }
     }
 
     }
 
-    @Override
-    public void close() {
-        disruptor.shutdown();
-        executor.shutdown();
+    private static class DefaultDOMEvent extends DefaultDOMNotification implements DOMEvent {
+        private final Date eventTime;
+
+        DefaultDOMEvent(org.opendaylight.mdsal.dom.api.DOMNotification fromNotification,
+                org.opendaylight.mdsal.dom.api.DOMEvent fromEvent) {
+            super(fromNotification);
+            final Instant eventInstant = fromEvent.getEventInstant();
+            this.eventTime = eventInstant != null ? Date.from(eventInstant) : null;
+        }
+
+        @Override
+        public Date getEventTime() {
+            return eventTime;
+        }
     }
 }
     }
 }
index 4cc85298eb40f5030a867b68159f6751508002e0..bff959c0edd8eaccf56cf2cd779c8c01c0ae2a0b 100644 (file)
 
   <!-- DOM Notification Service -->
 
 
   <!-- DOM Notification Service -->
 
+  <reference id="domNotificationService" interface="org.opendaylight.mdsal.dom.api.DOMNotificationService"
+      odl:type="default"/>
+  <reference id="domNotificationPublishService" interface="org.opendaylight.mdsal.dom.api.DOMNotificationPublishService"
+      odl:type="default"/>
+  <reference id="domNotificationSubscriptionListenerRegistry" interface="org.opendaylight.mdsal.dom.spi.DOMNotificationSubscriptionListenerRegistry"
+      odl:type="default"/>
+
   <bean id="domNotificationRouter" class="org.opendaylight.controller.md.sal.dom.broker.impl.DOMNotificationRouter"
           factory-method="create">
   <bean id="domNotificationRouter" class="org.opendaylight.controller.md.sal.dom.broker.impl.DOMNotificationRouter"
           factory-method="create">
-    <argument value="${notification-queue-depth}"/>
-    <argument value="${notification-queue-spin}"/>
-    <argument value="${notification-queue-park}"/>
-    <argument value="MILLISECONDS"/>
+    <argument ref="domNotificationService"/>
+    <argument ref="domNotificationPublishService"/>
+    <argument ref="domNotificationSubscriptionListenerRegistry"/>
   </bean>
 
   <service ref="domNotificationRouter" odl:type="default">
   </bean>
 
   <service ref="domNotificationRouter" odl:type="default">
diff --git a/opendaylight/md-sal/sal-dom-broker/src/test/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouterTest.java b/opendaylight/md-sal/sal-dom-broker/src/test/java/org/opendaylight/controller/md/sal/dom/broker/impl/DOMNotificationRouterTest.java
new file mode 100644 (file)
index 0000000..ab3a8a3
--- /dev/null
@@ -0,0 +1,269 @@
+/*
+ * Copyright (c) 2018 Inocybe Technologies 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.md.sal.dom.broker.impl;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import com.google.common.collect.ImmutableSet;
+import com.google.common.util.concurrent.SettableFuture;
+import com.google.common.util.concurrent.Uninterruptibles;
+import java.time.Instant;
+import java.util.Date;
+import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import javax.annotation.Nullable;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.opendaylight.controller.md.sal.dom.api.DOMEvent;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotification;
+import org.opendaylight.controller.md.sal.dom.api.DOMNotificationListener;
+import org.opendaylight.controller.md.sal.dom.spi.DOMNotificationSubscriptionListener;
+import org.opendaylight.controller.md.sal.dom.store.impl.TestModel;
+import org.opendaylight.yangtools.concepts.ListenerRegistration;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
+import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
+import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
+import org.opendaylight.yangtools.yang.data.impl.schema.builder.impl.ImmutableContainerNodeBuilder;
+import org.opendaylight.yangtools.yang.model.api.Module;
+import org.opendaylight.yangtools.yang.model.api.NotificationDefinition;
+import org.opendaylight.yangtools.yang.model.api.SchemaContext;
+import org.opendaylight.yangtools.yang.model.api.SchemaPath;
+
+/**
+ * Unit tests for DOMNotificationRouter.
+ *
+ * @author Thomas Pantelis
+ */
+public class DOMNotificationRouterTest {
+    private static final ContainerNode BODY = ImmutableContainerNodeBuilder.create().withNodeIdentifier(
+        new NodeIdentifier(QName.create(TestModel.TEST_QNAME.getModule(), "test-notification")))
+            .withChild(ImmutableNodes.leafNode(QName.create(TestModel.TEST_QNAME.getModule(), "value-leaf"), "foo"))
+                .build();
+    private static final Instant INSTANT = Instant.now();
+
+    private static SchemaPath notificationSchemaPath;
+
+    private final org.opendaylight.mdsal.dom.broker.DOMNotificationRouter mdsalRouter =
+            org.opendaylight.mdsal.dom.broker.DOMNotificationRouter.create(16);
+    private final DOMNotificationRouter legacyRouter =
+            DOMNotificationRouter.create(mdsalRouter, mdsalRouter, mdsalRouter);
+    private final TestLegacyDOMNotificationListener testLegacyListener = new TestLegacyDOMNotificationListener();
+    private final TestMdsalDOMNotificationListener testMdsalListener = new TestMdsalDOMNotificationListener();
+
+    @BeforeClass
+    public static void staticSetup() {
+        final SchemaContext schemaContext = TestModel.createTestContext();
+
+        Module testModule = schemaContext.findModule("odl-datastore-test", TestModel.TEST_QNAME.getRevision()).get();
+        NotificationDefinition notificationDefinition = null;
+        for (NotificationDefinition def: testModule.getNotifications()) {
+            if (def.getQName().getLocalName().equals("test-notification")) {
+                notificationDefinition = def;
+                break;
+            }
+        }
+
+        assertNotNull("test-notification not found in " + testModule.getNotifications(), notificationDefinition);
+        notificationSchemaPath = notificationDefinition.getPath();
+    }
+
+    @Test
+    public void testLegacyListenerAndPublish() throws InterruptedException, ExecutionException, TimeoutException {
+        final ListenerRegistration<TestLegacyDOMNotificationListener> reg =
+                legacyRouter.registerNotificationListener(testLegacyListener, notificationSchemaPath);
+
+        legacyRouter.putNotification(new TestLegacyDOMNotification()).get(5, TimeUnit.SECONDS);
+        testLegacyListener.verifyReceived(notificationSchemaPath, BODY, null);
+
+        legacyRouter.offerNotification(new TestLegacyDOMNotification()).get(5, TimeUnit.SECONDS);
+        testLegacyListener.verifyReceived(notificationSchemaPath, BODY, null);
+
+        legacyRouter.offerNotification(new TestLegacyDOMNotification(), 100, TimeUnit.MILLISECONDS)
+            .get(5, TimeUnit.SECONDS);
+        testLegacyListener.verifyReceived(notificationSchemaPath, BODY, null);
+
+        legacyRouter.offerNotification(new TestLegacyDOMEvent()).get(5, TimeUnit.SECONDS);
+        testLegacyListener.verifyReceived(notificationSchemaPath, BODY, Date.from(INSTANT));
+
+        reg.close();
+
+        legacyRouter.offerNotification(new TestLegacyDOMNotification()).get(5, TimeUnit.SECONDS);
+        testLegacyListener.verifyNotReceived();
+    }
+
+    @Test
+    public void testLegacyListenerAndMdsalPublish()
+            throws InterruptedException, ExecutionException, TimeoutException {
+        legacyRouter.registerNotificationListener(testLegacyListener, notificationSchemaPath);
+
+        mdsalRouter.offerNotification(new TestMdsalDOMNotification()).get(5, TimeUnit.SECONDS);
+        testLegacyListener.verifyReceived(notificationSchemaPath, BODY, null);
+
+        mdsalRouter.offerNotification(new TestMdsalDOMEvent()).get(5, TimeUnit.SECONDS);
+        testLegacyListener.verifyReceived(notificationSchemaPath, BODY, Date.from(INSTANT));
+    }
+
+    @Test
+    public void testMdsalListenerAndLegacyPublish()
+            throws InterruptedException, ExecutionException, TimeoutException {
+        mdsalRouter.registerNotificationListener(testMdsalListener, notificationSchemaPath);
+
+        legacyRouter.offerNotification(new TestLegacyDOMNotification()).get(5, TimeUnit.SECONDS);
+        testMdsalListener.verifyReceived(notificationSchemaPath, BODY, null);
+
+        legacyRouter.offerNotification(new TestLegacyDOMEvent()).get(5, TimeUnit.SECONDS);
+        testMdsalListener.verifyReceived(notificationSchemaPath, BODY, INSTANT);
+    }
+
+    @Test
+    public void testRegisterSubscriptionListener() throws InterruptedException, ExecutionException, TimeoutException {
+        TestLegacyDOMNotificationSubscriptionListener listener = new TestLegacyDOMNotificationSubscriptionListener();
+        final ListenerRegistration<TestLegacyDOMNotificationSubscriptionListener> subscriptionReg =
+                legacyRouter.registerSubscriptionListener(listener);
+
+        listener.verifyReceived();
+
+        final ListenerRegistration<TestLegacyDOMNotificationListener> listenerReg =
+                legacyRouter.registerNotificationListener(testLegacyListener, notificationSchemaPath);
+
+        listener.verifyReceived(notificationSchemaPath);
+
+        listenerReg.close();
+
+        listener.verifyReceived();
+
+        subscriptionReg.close();
+
+        legacyRouter.registerNotificationListener(testLegacyListener, notificationSchemaPath);
+
+        listener.verifyNotReceived();
+    }
+
+    private static class TestLegacyDOMNotificationListener implements DOMNotificationListener {
+        SettableFuture<DOMNotification> receivedNotification = SettableFuture.create();
+
+        @Override
+        public void onNotification(DOMNotification notification) {
+            receivedNotification.set(notification);
+        }
+
+        void verifyReceived(SchemaPath path, ContainerNode body, @Nullable Date eventTime)
+                throws InterruptedException, ExecutionException, TimeoutException {
+            final DOMNotification actual = receivedNotification.get(5, TimeUnit.SECONDS);
+            assertEquals(path, actual.getType());
+            assertEquals(body, actual.getBody());
+
+            if (eventTime != null) {
+                assertTrue("Expected DOMEvent", actual instanceof DOMEvent);
+                assertEquals(eventTime, ((DOMEvent)actual).getEventTime());
+            } else {
+                assertFalse("Unexpected DOMEvent", actual instanceof DOMEvent);
+            }
+
+            receivedNotification = SettableFuture.create();
+        }
+
+        void verifyNotReceived() {
+            Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
+            assertFalse("Unexpected notification", receivedNotification.isDone());
+        }
+    }
+
+    private static class TestMdsalDOMNotificationListener
+            implements org.opendaylight.mdsal.dom.api.DOMNotificationListener {
+        SettableFuture<org.opendaylight.mdsal.dom.api.DOMNotification> receivedNotification = SettableFuture.create();
+
+        @Override
+        public void onNotification(org.opendaylight.mdsal.dom.api.DOMNotification notification) {
+            receivedNotification.set(notification);
+        }
+
+        void verifyReceived(SchemaPath path, ContainerNode body, @Nullable Instant eventTime)
+                throws InterruptedException, ExecutionException, TimeoutException {
+            final org.opendaylight.mdsal.dom.api.DOMNotification actual =
+                    receivedNotification.get(5, TimeUnit.SECONDS);
+            assertEquals(path, actual.getType());
+            assertEquals(body, actual.getBody());
+
+            if (eventTime != null) {
+                assertTrue("Expected DOMEvent", actual instanceof org.opendaylight.mdsal.dom.api.DOMEvent);
+                assertEquals(eventTime, ((org.opendaylight.mdsal.dom.api.DOMEvent)actual).getEventInstant());
+            } else {
+                assertFalse("Unexpected DOMEvent", actual instanceof org.opendaylight.mdsal.dom.api.DOMEvent);
+            }
+
+            receivedNotification = SettableFuture.create();
+        }
+    }
+
+    private static class TestLegacyDOMNotificationSubscriptionListener implements DOMNotificationSubscriptionListener {
+        SettableFuture<Set<SchemaPath>> receivedNotification = SettableFuture.create();
+
+        @Override
+        public void onSubscriptionChanged(Set<SchemaPath> currentTypes) {
+            receivedNotification.set(currentTypes);
+        }
+
+        void verifyReceived(SchemaPath... paths)
+                throws InterruptedException, ExecutionException, TimeoutException {
+            final Set<SchemaPath> actual = receivedNotification.get(5, TimeUnit.SECONDS);
+            assertEquals(ImmutableSet.copyOf(paths), actual);
+            receivedNotification = SettableFuture.create();
+        }
+
+        void verifyNotReceived() {
+            Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
+            assertFalse("Unexpected notification", receivedNotification.isDone());
+        }
+    }
+
+    private static class TestLegacyDOMNotification implements DOMNotification {
+        @Override
+        public SchemaPath getType() {
+            return notificationSchemaPath;
+        }
+
+        @Override
+        public ContainerNode getBody() {
+            return BODY;
+        }
+    }
+
+    private static class TestLegacyDOMEvent extends TestLegacyDOMNotification implements DOMEvent {
+        @Override
+        public Date getEventTime() {
+            return Date.from(INSTANT);
+        }
+    }
+
+    private static class TestMdsalDOMNotification implements org.opendaylight.mdsal.dom.api.DOMNotification {
+        @Override
+        public SchemaPath getType() {
+            return notificationSchemaPath;
+        }
+
+        @Override
+        public ContainerNode getBody() {
+            return BODY;
+        }
+    }
+
+    private static class TestMdsalDOMEvent extends TestMdsalDOMNotification
+            implements org.opendaylight.mdsal.dom.api.DOMEvent {
+        @Override
+        public Instant getEventInstant() {
+            return INSTANT;
+        }
+    }
+}
index 802bb6f199de98cd2654c872b3d0d7170bef688a..d16c153f6a8b9cd36818e593395cd47ba312fb9c 100644 (file)
@@ -8,9 +8,7 @@
 package org.opendaylight.controller.md.sal.dom.spi;
 
 import com.google.common.annotations.Beta;
 package org.opendaylight.controller.md.sal.dom.spi;
 
 import com.google.common.annotations.Beta;
-import java.util.EventListener;
-import java.util.Set;
-import org.opendaylight.yangtools.yang.model.api.SchemaPath;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
 
 /**
  * Listener which is notified when subscriptions changes and
 
 /**
  * Listener which is notified when subscriptions changes and
@@ -19,12 +17,7 @@ import org.opendaylight.yangtools.yang.model.api.SchemaPath;
  *
  */
 @Beta
  *
  */
 @Beta
-public interface DOMNotificationSubscriptionListener extends EventListener {
-
-    /**
-     * Invoked when notification subscription changed.
-     *
-     * @param currentTypes Set of notification types for which listeners are registered.
-     */
-    void onSubscriptionChanged(Set<SchemaPath> currentTypes);
+@SuppressFBWarnings(value = "NM_SAME_SIMPLE_NAME_AS_INTERFACE", justification = "Migration")
+public interface DOMNotificationSubscriptionListener
+        extends org.opendaylight.mdsal.dom.spi.DOMNotificationSubscriptionListener {
 }
 }