Refactor Register*ListenerReply classes

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / AbstractDataListenerSupport.java

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 a253b794db15871f69117f291be14ef4d1193a23..5a09b87da19f1e5b360761b1e64d05f00d793737 100644 (file)
--- 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
@@ -9,101 +9,121 @@ package org.opendaylight.controller.cluster.datastore;
 
 import akka.actor.ActorRef;
 import akka.actor.ActorSelection;
 
 import akka.actor.ActorRef;
 import akka.actor.ActorSelection;

-import com.google.common.base.Optional;

 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.EventListener;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.EventListener;

-import java.util.Map.Entry;
+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.controller.cluster.datastore.messages.EnableNotification;
 import org.opendaylight.controller.cluster.datastore.messages.ListenerRegistrationMessage;

-import org.opendaylight.yangtools.concepts.ListenerRegistration;
-import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
+import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeNotificationListenerReply;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

-abstract class AbstractDataListenerSupport<L extends EventListener, R extends ListenerRegistrationMessage,
-        D extends DelayedListenerRegistration<L, R>, LR extends ListenerRegistration<L>>
-                extends LeaderLocalDelegateFactory<R, LR, Optional<DataTreeCandidate>> {
+abstract class AbstractDataListenerSupport<L extends EventListener, M extends ListenerRegistrationMessage>
+        extends LeaderLocalDelegateFactory<M> {

     private final Logger log = LoggerFactory.getLogger(getClass());
 
     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<DelayedListenerRegistration<L, M>> delayedListenerRegistrations =
+            ConcurrentHashMap.newKeySet();
+    private final Collection<DelayedListenerRegistration<L, M>> delayedListenerOnAllRegistrations =
+            ConcurrentHashMap.newKeySet();
+    private final Collection<ActorSelection> leaderOnlyListenerActors = ConcurrentHashMap.newKeySet();
+    private final Collection<ActorSelection> allListenerActors = ConcurrentHashMap.newKeySet();

 
     protected AbstractDataListenerSupport(Shard shard) {
         super(shard);
     }
 
 
     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);
 
     @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) {
+        final EnableNotification msg = new EnableNotification(isLeader, persistenceId());
+        for (ActorSelection dataChangeListener : leaderOnlyListenerActors) {

             dataChangeListener.tell(msg, getSelf());
         }
 
             dataChangeListener.tell(msg, getSelf());
         }
 

-        if(hasLeader) {
-            for(D reg : delayedListenerOnAllRegistrations) {
-                reg.createDelegate(this);
+        if (hasLeader) {
+            for (DelayedListenerRegistration<L, M> reg : delayedListenerOnAllRegistrations) {
+                reg.doRegistration(this);

             }
 
             delayedListenerOnAllRegistrations.clear();
             }
 
             delayedListenerOnAllRegistrations.clear();

-            delayedListenerOnAllRegistrations.trimToSize();

         }
 
         }
 

-        if(isLeader) {
-            for(D reg : delayedListenerRegistrations) {
-                reg.createDelegate(this);
+        if (isLeader) {
+            for (DelayedListenerRegistration<L, M> reg : delayedListenerRegistrations) {
+                reg.doRegistration(this);

             }
 
             delayedListenerRegistrations.clear();
             }
 
             delayedListenerRegistrations.clear();

-            delayedListenerRegistrations.trimToSize();

         }
     }
 
     @Override
         }
     }
 
     @Override

-    void onMessage(R message, boolean isLeader, boolean hasLeader) {
-        log.debug("{}: {} for {}, leader: {}", persistenceId(), logName(), message.getPath(), isLeader);
+    void onMessage(M message, boolean isLeader, boolean hasLeader) {
+        log.debug("{}: {} for {}, isLeader: {}, hasLeader: {}", persistenceId(), logName(), message,
+                isLeader, hasLeader);

 
 

-        final ListenerRegistration<L> registration;
-        if((hasLeader && message.isRegisterOnAllInstances()) || isLeader) {
-            final Entry<LR, Optional<DataTreeCandidate>> res = createDelegate(message);
-            registration = res.getKey();
+        ActorRef registrationActor = createActor(DataTreeNotificationListenerRegistrationActor.props());
+
+        if (hasLeader && message.isRegisterOnAllInstances() || isLeader) {
+            doRegistration(message, registrationActor);

         } else {
             log.debug("{}: Shard is not the leader - delaying registration", persistenceId());
 
         } else {
             log.debug("{}: Shard is not the leader - delaying registration", persistenceId());
 

-            D delayedReg = newDelayedListenerRegistration(message);
-            if(message.isRegisterOnAllInstances()) {
-                delayedListenerOnAllRegistrations.add(delayedReg);
+            DelayedListenerRegistration<L, M> delayedReg =
+                    new DelayedListenerRegistration<>(message, registrationActor);
+            Collection<DelayedListenerRegistration<L, M>> delayedRegList;
+            if (message.isRegisterOnAllInstances()) {
+                delayedRegList = delayedListenerOnAllRegistrations;

             } else {
             } 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());
 
         log.debug("{}: {} sending reply, listenerRegistrationPath = {} ", persistenceId(), logName(),
                 registrationActor.path());
 

-        tellSender(newRegistrationReplyMessage(registrationActor));
+        tellSender(new RegisterDataTreeNotificationListenerReply(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, persistenceId()), 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 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(R message);
-
-    protected abstract ActorRef newRegistrationActor(ListenerRegistration<L> registration);
-
-    protected abstract Object newRegistrationReplyMessage(ActorRef registrationActor);
+    abstract void doRegistration(M message, ActorRef registrationActor);

 
     protected abstract String logName();
 }
 
     protected abstract String logName();
 }