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