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%2FDataTreeChangeListenerSupport.java;h=4f70327cb7246e947801902c05a39d2f5c81cc3b;hp=3987c9af359a31ec7dcde5c2fa6e33911ed2a19d;hb=ec870dee9bacb971f11bc747b69e84ac37f5d746;hpb=c389b6dc9c717695f88a83c01a9fc67a1df68aac diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerSupport.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerSupport.java index 3987c9af35..4f70327cb7 100644 --- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerSupport.java +++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerSupport.java @@ -9,82 +9,46 @@ package org.opendaylight.controller.cluster.datastore; import akka.actor.ActorRef; import akka.actor.ActorSelection; -import java.util.ArrayList; -import java.util.Collection; -import org.opendaylight.controller.cluster.datastore.messages.EnableNotification; +import org.opendaylight.controller.cluster.datastore.actors.DataTreeNotificationListenerRegistrationActor; import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener; import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListenerReply; import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener; -import org.opendaylight.yangtools.concepts.ListenerRegistration; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -final class DataTreeChangeListenerSupport extends LeaderLocalDelegateFactory> { - private static final Logger LOG = LoggerFactory.getLogger(DataTreeChangeListenerSupport.class); - private final ArrayList delayedRegistrations = new ArrayList<>(); - private final Collection actors = new ArrayList<>(); +final class DataTreeChangeListenerSupport extends AbstractDataListenerSupport { DataTreeChangeListenerSupport(final Shard shard) { super(shard); } @Override - void onLeadershipChange(final boolean isLeader) { - if (isLeader) { - for (DelayedDataTreeListenerRegistration reg : delayedRegistrations) { - reg.createDelegate(this); - } - delayedRegistrations.clear(); - delayedRegistrations.trimToSize(); - } + void doRegistration(final RegisterDataTreeChangeListener message, final ActorRef registrationActor) { + final ActorSelection listenerActor = processListenerRegistrationMessage(message); - final EnableNotification msg = new EnableNotification(isLeader); - for (ActorSelection dataChangeListener : actors) { - dataChangeListener.tell(msg, getSelf()); - } - } - - @Override - void onMessage(final RegisterDataTreeChangeListener registerTreeChangeListener, final boolean isLeader) { - LOG.debug("{}: registerTreeChangeListener for {}, leader: {}", persistenceId(), registerTreeChangeListener.getPath(), isLeader); - - final ListenerRegistration registration; - if (!isLeader) { - LOG.debug("{}: Shard is not the leader - delaying registration", persistenceId()); + DOMDataTreeChangeListener listener = new ForwardingDataTreeChangeListener(listenerActor); - DelayedDataTreeListenerRegistration delayedReg = - new DelayedDataTreeListenerRegistration(registerTreeChangeListener); - delayedRegistrations.add(delayedReg); - registration = delayedReg; - } else { - registration = createDelegate(registerTreeChangeListener); - } + log().debug("{}: Registering for path {}", persistenceId(), message.getPath()); - ActorRef listenerRegistration = createActor(DataTreeChangeListenerRegistrationActor.props(registration)); - - LOG.debug("{}: registerDataChangeListener sending reply, listenerRegistrationPath = {} ", - persistenceId(), listenerRegistration.path()); - - tellSender(new RegisterDataTreeChangeListenerReply(listenerRegistration)); + final ShardDataTree shardDataTree = getShard().getDataStore(); + shardDataTree.registerTreeChangeListener(message.getPath(), + listener, shardDataTree.readCurrentData(), registration -> registrationActor.tell( + new DataTreeNotificationListenerRegistrationActor.SetRegistration(registration, () -> + removeListenerActor(listenerActor)), ActorRef.noSender())); } @Override - ListenerRegistration createDelegate(final RegisterDataTreeChangeListener message) { - ActorSelection dataChangeListenerPath = selectActor(message.getDataTreeChangeListenerPath()); - - // Notify the listener if notifications should be enabled or not - // If this shard is the leader then it will enable notifications else - // it will not - dataChangeListenerPath.tell(new EnableNotification(true), getSelf()); - - // Now store a reference to the data change listener so it can be notified - // at a later point if notifications should be enabled or disabled - actors.add(dataChangeListenerPath); - - DOMDataTreeChangeListener listener = new ForwardingDataTreeChangeListener(dataChangeListenerPath); + protected DelayedDataTreeListenerRegistration newDelayedListenerRegistration( + RegisterDataTreeChangeListener message, ActorRef registrationActor) { + return new DelayedDataTreeListenerRegistration(message, registrationActor); + } - LOG.debug("{}: Registering for path {}", persistenceId(), message.getPath()); + @Override + protected Object newRegistrationReplyMessage(ActorRef registrationActor) { + return new RegisterDataTreeChangeListenerReply(registrationActor); + } - return getShard().getDataStore().registerTreeChangeListener(message.getPath(), listener); + @Override + protected String logName() { + return "registerTreeChangeListener"; } }