X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=dom%2Fmdsal-dom-inmemory-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fmdsal%2Fdom%2Fstore%2Finmemory%2FInMemoryDOMStoreTreeChangePublisher.java;h=87b6612bd3c0f8ecbf2a3ce771230acc5969209b;hb=HEAD;hp=c92f40b4ae16923d5b041c43a03700bf6c234dac;hpb=0f351bbc28ddf2cddfe30c8d018646d81953fa17;p=mdsal.git

diff --git a/dom/mdsal-dom-inmemory-datastore/src/main/java/org/opendaylight/mdsal/dom/store/inmemory/InMemoryDOMStoreTreeChangePublisher.java b/dom/mdsal-dom-inmemory-datastore/src/main/java/org/opendaylight/mdsal/dom/store/inmemory/InMemoryDOMStoreTreeChangePublisher.java
index c92f40b4ae..87b6612bd3 100644
--- a/dom/mdsal-dom-inmemory-datastore/src/main/java/org/opendaylight/mdsal/dom/store/inmemory/InMemoryDOMStoreTreeChangePublisher.java
+++ b/dom/mdsal-dom-inmemory-datastore/src/main/java/org/opendaylight/mdsal/dom/store/inmemory/InMemoryDOMStoreTreeChangePublisher.java
@@ -7,87 +7,97 @@
  */
 package org.opendaylight.mdsal.dom.store.inmemory;
 
-import com.google.common.collect.ImmutableList;
-import java.util.Collection;
-import java.util.Optional;
-import java.util.concurrent.ExecutorService;
-import javax.annotation.Nonnull;
+import java.util.List;
+import java.util.concurrent.Executor;
+import org.eclipse.jdt.annotation.NonNull;
 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeListener;
-import org.opendaylight.mdsal.dom.spi.AbstractDOMDataTreeChangeListenerRegistration;
 import org.opendaylight.mdsal.dom.spi.store.AbstractDOMStoreTreeChangePublisher;
-import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager;
-import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager.BatchedInvoker;
+import org.opendaylight.yangtools.concepts.Registration;
+import org.opendaylight.yangtools.util.concurrent.EqualityQueuedNotificationManager;
 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
-import org.opendaylight.yangtools.yang.data.api.schema.NormalizedNode;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidates;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeSnapshot;
+import org.opendaylight.yangtools.yang.data.api.schema.DataContainerNode;
+import org.opendaylight.yangtools.yang.data.tree.api.DataTreeCandidate;
+import org.opendaylight.yangtools.yang.data.tree.api.DataTreeSnapshot;
+import org.opendaylight.yangtools.yang.data.tree.spi.DataTreeCandidates;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 final class InMemoryDOMStoreTreeChangePublisher extends AbstractDOMStoreTreeChangePublisher {
-    private static final BatchedInvoker<AbstractDOMDataTreeChangeListenerRegistration<?>, DataTreeCandidate>
-        MANAGER_INVOKER = (listener, notifications) -> {
-            final DOMDataTreeChangeListener inst = listener.getInstance();
-            if (inst != null) {
-                inst.onDataTreeChanged(ImmutableList.copyOf(notifications));
-            }
-        };
     private static final Logger LOG = LoggerFactory.getLogger(InMemoryDOMStoreTreeChangePublisher.class);
 
-    private final QueuedNotificationManager<AbstractDOMDataTreeChangeListenerRegistration<?>, DataTreeCandidate>
-        notificationManager;
+    // Registrations use identity for equality, hence we can skip wrapping them
+    private final EqualityQueuedNotificationManager<Reg, DataTreeCandidate> notificationManager;
 
-    InMemoryDOMStoreTreeChangePublisher(final ExecutorService listenerExecutor, final int maxQueueSize) {
-        notificationManager = QueuedNotificationManager.create(listenerExecutor, MANAGER_INVOKER, maxQueueSize,
-                "DataTreeChangeListenerQueueMgr");
+    InMemoryDOMStoreTreeChangePublisher(final String dsName, final Executor listenerExecutor, final int maxQueueSize) {
+        notificationManager = new EqualityQueuedNotificationManager<>("DataTreeChangeListenerQueueMgr + dsName",
+            listenerExecutor, maxQueueSize,
+            (listener, notifications) -> {
+                if (listener.notClosed()) {
+                    listener.listener().onDataTreeChanged(notifications);
+                }
+            });
     }
 
-    private InMemoryDOMStoreTreeChangePublisher(final QueuedNotificationManager<
-            AbstractDOMDataTreeChangeListenerRegistration<?>, DataTreeCandidate> notificationManager) {
+    private InMemoryDOMStoreTreeChangePublisher(
+            final EqualityQueuedNotificationManager<Reg, DataTreeCandidate> notificationManager) {
         this.notificationManager = notificationManager;
     }
 
-    QueuedNotificationManager<?, ?> getNotificationManager() {
+    EqualityQueuedNotificationManager<?, ?> getNotificationManager() {
         return notificationManager;
     }
 
     @Override
-    protected void notifyListener(final AbstractDOMDataTreeChangeListenerRegistration<?> registration,
-            final Collection<DataTreeCandidate> changes) {
+    protected void notifyListener(final Reg registration, final List<DataTreeCandidate> changes) {
         LOG.debug("Enqueueing candidates {} for registration {}", changes, registration);
         notificationManager.submitNotifications(registration, changes);
     }
 
     @Override
-    protected synchronized void registrationRemoved(
-            final AbstractDOMDataTreeChangeListenerRegistration<?> registration) {
+    protected synchronized void registrationRemoved(final Reg registration) {
         LOG.debug("Closing registration {}", registration);
 
         // FIXME: remove the queue for this registration and make sure we clear it
     }
 
-    <L extends DOMDataTreeChangeListener> ListenerRegistration<L> registerTreeChangeListener(
-            final YangInstanceIdentifier treeId, final L listener, final DataTreeSnapshot snapshot) {
-        final AbstractDOMDataTreeChangeListenerRegistration<L> reg = registerTreeChangeListener(treeId, listener);
+    Registration registerTreeChangeListener(final YangInstanceIdentifier treeId,
+            final DOMDataTreeChangeListener listener, final DataTreeSnapshot snapshot) {
+        final var reg = registerTreeChangeListener(treeId, listener);
+        final var preExistingData = snapshot.readNode(YangInstanceIdentifier.of());
+        if (preExistingData.isEmpty()) {
+            listener.onInitialData();
+            return reg;
+        }
 
-        final Optional<NormalizedNode<?, ?>> node = snapshot.readNode(YangInstanceIdentifier.EMPTY);
-        if (node.isPresent()) {
-            final DataTreeCandidate candidate = DataTreeCandidates.fromNormalizedNode(
-                    YangInstanceIdentifier.EMPTY, node.get());
+        final var data = preExistingData.orElseThrow();
+        if (treeId.isEmpty()) {
+            if (data instanceof DataContainerNode container) {
+                if (container.isEmpty()) {
+                    // If we are listening on root of data tree we still get empty normalized node, root is always
+                    // present, we should filter this out separately and notify it by 'onInitialData()' once.
+                    // Otherwise, it is just a valid data node with empty value which also should be notified by
+                    // "onDataTreeChanged(List<DataTreeCandidate>)".
+                    listener.onInitialData();
+                    return reg;
+                }
+            } else {
+                throw new IllegalStateException("Unexpected root node type " + data.contract());
+            }
+        }
 
-            InMemoryDOMStoreTreeChangePublisher publisher =
-                    new InMemoryDOMStoreTreeChangePublisher(notificationManager);
-            publisher.registerTreeChangeListener(treeId, listener);
-            publisher.publishChange(candidate);
+        final var candidate = DataTreeCandidates.fromNormalizedNode(YangInstanceIdentifier.of(), data);
+        final var publisher = new InMemoryDOMStoreTreeChangePublisher(notificationManager);
+        publisher.registerTreeChangeListener(treeId, listener);
+        if (!publisher.publishChange(candidate)) {
+            // There is no data in the conceptual data tree then notify with 'onInitialData()'.
+            listener.onInitialData();
         }
 
         return reg;
     }
 
-    synchronized void publishChange(@Nonnull final DataTreeCandidate candidate) {
+    synchronized boolean publishChange(final @NonNull DataTreeCandidate candidate) {
         // Runs synchronized with registrationRemoved()
-        processCandidateTree(candidate);
+        return processCandidateTree(candidate);
     }
 }