opendaylight/md-sal/sal-clustering-commons/src/main/java/org/opendaylight/controller/cluster/notifications/RoleChangeNotifier.java

   1 /*
   2  * Copyright (c) 2014 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.notifications;
   9
  10 import akka.actor.ActorPath;
  11 import akka.actor.ActorRef;
  12 import akka.actor.Props;
  13 import akka.serialization.Serialization;
  14 import java.util.HashMap;
  15 import java.util.Map;
  16 import org.opendaylight.controller.cluster.common.actor.AbstractUntypedActor;
  17
  18 /**
  19  * The RoleChangeNotifier is responsible for receiving Raft role and leader state change messages and notifying
  20  * the listeners (within the same node), which are registered with it.
  21  *
  22  * <p>
  23  * The RoleChangeNotifier is instantiated by the Shard and injected into the RaftActor.
  24  */
  25 public class RoleChangeNotifier extends AbstractUntypedActor implements AutoCloseable {
  26
  27     private final String memberId;
  28     private final Map<ActorPath, ActorRef> registeredListeners = new HashMap<>();
  29     private RoleChangeNotification latestRoleChangeNotification = null;
  30     private LeaderStateChanged latestLeaderStateChanged;
  31
  32     public RoleChangeNotifier(final String memberId) {
  33         this.memberId = memberId;
  34     }
  35
  36     public static Props getProps(final String memberId) {
  37         return Props.create(RoleChangeNotifier.class, memberId);
  38     }
  39
  40     @Override
  41     public void preStart() throws Exception {
  42         super.preStart();
  43         LOG.info("RoleChangeNotifier:{} created and ready for shard:{}",
  44             Serialization.serializedActorPath(getSelf()), memberId);
  45     }
  46
  47     @Override
  48     protected void handleReceive(final Object message) {
  49         if (message instanceof RegisterRoleChangeListener) {
  50             // register listeners for this shard
  51
  52             ActorRef curRef = registeredListeners.get(getSender().path());
  53             if (curRef != null) {
  54                 // ActorPaths would pass equal even if the unique id of the actors are different
  55                 // if a listener actor is re-registering after reincarnation, then removing the existing
  56                 // entry so the actor path with correct unique id is registered.
  57                 registeredListeners.remove(getSender().path());
  58             }
  59             registeredListeners.put(getSender().path(), getSender());
  60
  61             LOG.info("RoleChangeNotifier for {} , registered listener {}", memberId,
  62                 getSender().path().toString());
  63
  64             getSender().tell(new RegisterRoleChangeListenerReply(), getSelf());
  65
  66             if (latestLeaderStateChanged != null) {
  67                 getSender().tell(latestLeaderStateChanged, getSelf());
  68             }
  69
  70             if (latestRoleChangeNotification != null) {
  71                 getSender().tell(latestRoleChangeNotification, getSelf());
  72             }
  73
  74
  75         } else if (message instanceof RoleChanged roleChanged) {
  76             // this message is sent by RaftActor. Notify registered listeners when this message is received.
  77
  78             LOG.info("RoleChangeNotifier for {} , received role change from {} to {}", memberId,
  79                 roleChanged.getOldRole(), roleChanged.getNewRole());
  80
  81             latestRoleChangeNotification =
  82                 new RoleChangeNotification(roleChanged.getMemberId(),
  83                     roleChanged.getOldRole(), roleChanged.getNewRole());
  84
  85             for (ActorRef listener: registeredListeners.values()) {
  86                 listener.tell(latestRoleChangeNotification, getSelf());
  87             }
  88         } else if (message instanceof LeaderStateChanged leaderStateChanged) {
  89             latestLeaderStateChanged = leaderStateChanged;
  90
  91             for (ActorRef listener : registeredListeners.values()) {
  92                 listener.tell(latestLeaderStateChanged, getSelf());
  93             }
  94         } else {
  95             unknownMessage(message);
  96         }
  97     }
  98
  99     @Override
 100     public void close() {
 101         registeredListeners.clear();
 102     }
 103 }
 104