X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatastore%2FAbstractDataListenerSupport.java;h=ac7a9337d9ff0a28507d07f75d199a1f1ed82b4c;hp=0821951a1a67b7cc30d7c98dae86994119af5cf6;hb=ec870dee9bacb971f11bc747b69e84ac37f5d746;hpb=e7512222d7d9e3149feb6a90eeb726e9391887fa

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataListenerSupport.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataListenerSupport.java
index 0821951a1a..ac7a9337d9 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataListenerSupport.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/AbstractDataListenerSupport.java
@@ -12,31 +12,36 @@ import akka.actor.ActorSelection;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.EventListener;
+import java.util.concurrent.ConcurrentHashMap;
+import org.opendaylight.controller.cluster.datastore.actors.DataTreeNotificationListenerRegistrationActor;
 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
 import org.opendaylight.controller.cluster.datastore.messages.ListenerRegistrationMessage;
-import org.opendaylight.yangtools.concepts.ListenerRegistration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 abstract class AbstractDataListenerSupport<L extends EventListener, M extends ListenerRegistrationMessage,
-        D extends DelayedListenerRegistration<L, M>, R extends ListenerRegistration<L>>
-                extends LeaderLocalDelegateFactory<M, R> {
+        D extends DelayedListenerRegistration<L, M>> extends LeaderLocalDelegateFactory<M> {
     private final Logger log = LoggerFactory.getLogger(getClass());
 
-    private final ArrayList<D> delayedListenerRegistrations = new ArrayList<>();
-    private final ArrayList<D> delayedListenerOnAllRegistrations = new ArrayList<>();
-    private final Collection<ActorSelection> actors = new ArrayList<>();
+    private final Collection<D> delayedListenerRegistrations = ConcurrentHashMap.newKeySet();
+    private final Collection<D> delayedListenerOnAllRegistrations = ConcurrentHashMap.newKeySet();
+    private final Collection<ActorSelection> leaderOnlyListenerActors = ConcurrentHashMap.newKeySet();
+    private final Collection<ActorSelection> allListenerActors = ConcurrentHashMap.newKeySet();
 
     protected AbstractDataListenerSupport(Shard shard) {
         super(shard);
     }
 
+    Collection<ActorSelection> getListenerActors() {
+        return new ArrayList<>(allListenerActors);
+    }
+
     @Override
     void onLeadershipChange(boolean isLeader, boolean hasLeader) {
         log.debug("{}: onLeadershipChange, isLeader: {}, hasLeader : {}", persistenceId(), isLeader, hasLeader);
 
         final EnableNotification msg = new EnableNotification(isLeader);
-        for (ActorSelection dataChangeListener : actors) {
+        for (ActorSelection dataChangeListener : leaderOnlyListenerActors) {
             dataChangeListener.tell(msg, getSelf());
         }
 
@@ -46,7 +51,6 @@ abstract class AbstractDataListenerSupport<L extends EventListener, M extends Li
             }
 
             delayedListenerOnAllRegistrations.clear();
-            delayedListenerOnAllRegistrations.trimToSize();
         }
 
         if (isLeader) {
@@ -55,7 +59,6 @@ abstract class AbstractDataListenerSupport<L extends EventListener, M extends Li
             }
 
             delayedListenerRegistrations.clear();
-            delayedListenerRegistrations.trimToSize();
         }
     }
 
@@ -63,41 +66,61 @@ abstract class AbstractDataListenerSupport<L extends EventListener, M extends Li
     void onMessage(M message, boolean isLeader, boolean hasLeader) {
         log.debug("{}: {} for {}, leader: {}", persistenceId(), logName(), message.getPath(), isLeader);
 
-        final ListenerRegistration<L> registration;
+        ActorRef registrationActor = createActor(DataTreeNotificationListenerRegistrationActor.props());
+
         if (hasLeader && message.isRegisterOnAllInstances() || isLeader) {
-            registration = createDelegate(message);
+            doRegistration(message, registrationActor);
         } else {
             log.debug("{}: Shard is not the leader - delaying registration", persistenceId());
 
-            D delayedReg = newDelayedListenerRegistration(message);
+            D delayedReg = newDelayedListenerRegistration(message, registrationActor);
+            Collection<D> delayedRegList;
             if (message.isRegisterOnAllInstances()) {
-                delayedListenerOnAllRegistrations.add(delayedReg);
+                delayedRegList = delayedListenerOnAllRegistrations;
             } else {
-                delayedListenerRegistrations.add(delayedReg);
+                delayedRegList = delayedListenerRegistrations;
             }
 
-            registration = delayedReg;
+            delayedRegList.add(delayedReg);
+            registrationActor.tell(new DataTreeNotificationListenerRegistrationActor.SetRegistration(
+                    delayedReg, () -> delayedRegList.remove(delayedReg)), ActorRef.noSender());
         }
 
-        ActorRef registrationActor = newRegistrationActor(registration);
-
         log.debug("{}: {} sending reply, listenerRegistrationPath = {} ", persistenceId(), logName(),
                 registrationActor.path());
 
         tellSender(newRegistrationReplyMessage(registrationActor));
     }
 
+    protected ActorSelection processListenerRegistrationMessage(M message) {
+        final ActorSelection listenerActor = selectActor(message.getListenerActorPath());
+
+        // We have a leader so enable the listener.
+        listenerActor.tell(new EnableNotification(true), getSelf());
+
+        if (!message.isRegisterOnAllInstances()) {
+            // This is a leader-only registration so store a reference to the listener actor so it can be notified
+            // at a later point if notifications should be enabled or disabled.
+            leaderOnlyListenerActors.add(listenerActor);
+        }
+
+        allListenerActors.add(listenerActor);
+
+        return listenerActor;
+    }
+
     protected Logger log() {
         return log;
     }
 
-    protected void addListenerActor(ActorSelection actor) {
-        actors.add(actor);
+    protected void removeListenerActor(ActorSelection listenerActor) {
+        allListenerActors.remove(listenerActor);
+        leaderOnlyListenerActors.remove(listenerActor);
     }
 
-    protected abstract D newDelayedListenerRegistration(M message);
+    abstract void doRegistration(M message, ActorRef registrationActor);
 
-    protected abstract ActorRef newRegistrationActor(ListenerRegistration<L> registration);
+    protected abstract D newDelayedListenerRegistration(M message, ActorRef registrationActor);
 
     protected abstract Object newRegistrationReplyMessage(ActorRef registrationActor);