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

   1 /*
   2  * Copyright (c) 2015 Brocade Communications 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.Optional;
  13 import java.util.ArrayList;
  14 import java.util.Collection;
  15 import java.util.EventListener;
  16 import java.util.Map.Entry;
  17 import org.opendaylight.controller.cluster.datastore.messages.EnableNotification;
  18 import org.opendaylight.controller.cluster.datastore.messages.ListenerRegistrationMessage;
  19 import org.opendaylight.yangtools.concepts.ListenerRegistration;
  20 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
  21 import org.slf4j.Logger;
  22 import org.slf4j.LoggerFactory;
  23
  24 abstract class AbstractDataListenerSupport<L extends EventListener, R extends ListenerRegistrationMessage,
  25         D extends DelayedListenerRegistration<L, R>, LR extends ListenerRegistration<L>>
  26                 extends LeaderLocalDelegateFactory<R, LR, Optional<DataTreeCandidate>> {
  27     private final Logger log = LoggerFactory.getLogger(getClass());
  28
  29     private final ArrayList<D> delayedListenerRegistrations = new ArrayList<>();
  30     private final ArrayList<D> delayedListenerOnAllRegistrations = new ArrayList<>();
  31     private final Collection<ActorSelection> actors = new ArrayList<>();
  32
  33     protected AbstractDataListenerSupport(Shard shard) {
  34         super(shard);
  35     }
  36
  37     @Override
  38     void onLeadershipChange(boolean isLeader, boolean hasLeader) {
  39         log.debug("{}: onLeadershipChange, isLeader: {}, hasLeader : {}", persistenceId(), isLeader, hasLeader);
  40
  41         final EnableNotification msg = new EnableNotification(isLeader);
  42         for(ActorSelection dataChangeListener : actors) {
  43             dataChangeListener.tell(msg, getSelf());
  44         }
  45
  46         if(hasLeader) {
  47             for(D reg : delayedListenerOnAllRegistrations) {
  48                 reg.createDelegate(this);
  49             }
  50
  51             delayedListenerOnAllRegistrations.clear();
  52             delayedListenerOnAllRegistrations.trimToSize();
  53         }
  54
  55         if(isLeader) {
  56             for(D reg : delayedListenerRegistrations) {
  57                 reg.createDelegate(this);
  58             }
  59
  60             delayedListenerRegistrations.clear();
  61             delayedListenerRegistrations.trimToSize();
  62         }
  63     }
  64
  65     @Override
  66     void onMessage(R message, boolean isLeader, boolean hasLeader) {
  67         log.debug("{}: {} for {}, leader: {}", persistenceId(), logName(), message.getPath(), isLeader);
  68
  69         final ListenerRegistration<L> registration;
  70         if((hasLeader && message.isRegisterOnAllInstances()) || isLeader) {
  71             final Entry<LR, Optional<DataTreeCandidate>> res = createDelegate(message);
  72             registration = res.getKey();
  73         } else {
  74             log.debug("{}: Shard is not the leader - delaying registration", persistenceId());
  75
  76             D delayedReg = newDelayedListenerRegistration(message);
  77             if(message.isRegisterOnAllInstances()) {
  78                 delayedListenerOnAllRegistrations.add(delayedReg);
  79             } else {
  80                 delayedListenerRegistrations.add(delayedReg);
  81             }
  82
  83             registration = delayedReg;
  84         }
  85
  86         ActorRef registrationActor = newRegistrationActor(registration);
  87
  88         log.debug("{}: {} sending reply, listenerRegistrationPath = {} ", persistenceId(), logName(),
  89                 registrationActor.path());
  90
  91         tellSender(newRegistrationReplyMessage(registrationActor));
  92     }
  93
  94     protected Logger log() {
  95         return log;
  96     }
  97
  98     protected void addListenerActor(ActorSelection actor) {
  99         actors.add(actor);
 100     }
 101
 102     protected abstract D newDelayedListenerRegistration(R message);
 103
 104     protected abstract ActorRef newRegistrationActor(ListenerRegistration<L> registration);
 105
 106     protected abstract Object newRegistrationReplyMessage(ActorRef registrationActor);
 107
 108     protected abstract String logName();
 109 }