opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/DataTreeChangeListenerSupport.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.datastore;
   9
  10 import akka.actor.ActorRef;
  11 import akka.actor.ActorSelection;
  12 import com.google.common.base.Preconditions;
  13 import java.util.ArrayList;
  14 import java.util.Collection;
  15 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
  16 import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListener;
  17 import org.opendaylight.controller.cluster.datastore.messages.RegisterDataTreeChangeListenerReply;
  18 import org.opendaylight.controller.md.sal.dom.api.DOMDataTreeChangeListener;
  19 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  20 import org.slf4j.Logger;
  21 import org.slf4j.LoggerFactory;
  22
  23 final class DataTreeChangeListenerSupport extends LeaderLocalDelegateFactory<RegisterDataTreeChangeListener, ListenerRegistration<DOMDataTreeChangeListener>> {
  24     private static final Logger LOG = LoggerFactory.getLogger(DataTreeChangeListenerSupport.class);
  25     private final ArrayList<DelayedDataTreeListenerRegistration> delayedRegistrations = new ArrayList<>();
  26     private final Collection<ActorSelection> actors = new ArrayList<>();
  27     private final Shard shard;
  28
  29     DataTreeChangeListenerSupport(final Shard shard) {
  30         this.shard = Preconditions.checkNotNull(shard);
  31     }
  32
  33     @Override
  34     void onLeadershipChange(final boolean isLeader) {
  35         if (isLeader) {
  36             for (DelayedDataTreeListenerRegistration reg : delayedRegistrations) {
  37                 reg.createDelegate(this);
  38             }
  39             delayedRegistrations.clear();
  40             delayedRegistrations.trimToSize();
  41         }
  42
  43         final EnableNotification msg = new EnableNotification(isLeader);
  44         for (ActorSelection dataChangeListener : actors) {
  45             dataChangeListener.tell(msg, shard.getSelf());
  46         }
  47     }
  48
  49     @Override
  50     void onMessage(final RegisterDataTreeChangeListener registerTreeChangeListener, final boolean isLeader) {
  51         LOG.debug("{}: registerTreeChangeListener for {}, leader: {}", shard.persistenceId(), registerTreeChangeListener.getPath(), isLeader);
  52
  53         final ListenerRegistration<DOMDataTreeChangeListener> registration;
  54         if (!isLeader) {
  55             LOG.debug("{}: Shard is not the leader - delaying registration", shard.persistenceId());
  56
  57             DelayedDataTreeListenerRegistration delayedReg =
  58                     new DelayedDataTreeListenerRegistration(registerTreeChangeListener);
  59             delayedRegistrations.add(delayedReg);
  60             registration = delayedReg;
  61         } else {
  62             registration = createDelegate(registerTreeChangeListener);
  63         }
  64
  65         ActorRef listenerRegistration = shard.getContext().actorOf(
  66                 DataTreeChangeListenerRegistrationActor.props(registration));
  67
  68         LOG.debug("{}: registerDataChangeListener sending reply, listenerRegistrationPath = {} ",
  69             shard.persistenceId(), listenerRegistration.path());
  70
  71         shard.getSender().tell(new RegisterDataTreeChangeListenerReply(listenerRegistration), shard.getSelf());
  72     }
  73
  74     @Override
  75     ListenerRegistration<DOMDataTreeChangeListener> createDelegate(final RegisterDataTreeChangeListener message) {
  76         ActorSelection dataChangeListenerPath = shard.getContext().system().actorSelection(
  77             message.getDataTreeChangeListenerPath().path());
  78
  79         // Notify the listener if notifications should be enabled or not
  80         // If this shard is the leader then it will enable notifications else
  81         // it will not
  82         dataChangeListenerPath.tell(new EnableNotification(true), shard.getSelf());
  83
  84         // Now store a reference to the data change listener so it can be notified
  85         // at a later point if notifications should be enabled or disabled
  86         actors.add(dataChangeListenerPath);
  87
  88         DOMDataTreeChangeListener listener = new ForwardingDataTreeChangeListener(dataChangeListenerPath);
  89
  90         LOG.debug("{}: Registering for path {}", shard.persistenceId(), message.getPath());
  91
  92         return shard.getDataStore().registerTreeChangeListener(message.getPath(), listener);
  93     }
  94 }