opendaylight/md-sal/sal-akka-raft/src/main/java/org/opendaylight/controller/cluster/raft/behaviors/Leader.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.raft.behaviors;
  10
  11 import akka.actor.ActorRef;
  12 import com.google.common.base.Preconditions;
  13 import org.opendaylight.controller.cluster.raft.FollowerLogInformationImpl;
  14 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  15 import org.opendaylight.controller.cluster.raft.RaftReplicator;
  16 import org.opendaylight.controller.cluster.raft.RaftState;
  17 import org.opendaylight.controller.cluster.raft.internal.messages.SendHeartBeat;
  18 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  19
  20 import java.util.Collections;
  21 import java.util.HashMap;
  22 import java.util.List;
  23 import java.util.Map;
  24 import java.util.concurrent.atomic.AtomicLong;
  25
  26 /**
  27  * The behavior of a RaftActor when it is in the Leader state
  28  * <p>
  29  * Leaders:
  30  * <ul>
  31  * <li> Upon election: send initial empty AppendEntries RPCs
  32  * (heartbeat) to each server; repeat during idle periods to
  33  * prevent election timeouts (§5.2)
  34  * <li> If command received from client: append entry to local log,
  35  * respond after entry applied to state machine (§5.3)
  36  * <li> If last log index ≥ nextIndex for a follower: send
  37  * AppendEntries RPC with log entries starting at nextIndex
  38  * <ul>
  39  * <li> If successful: update nextIndex and matchIndex for
  40  * follower (§5.3)
  41  * <li> If AppendEntries fails because of log inconsistency:
  42  * decrement nextIndex and retry (§5.3)
  43  * </ul>
  44  * <li> If there exists an N such that N > commitIndex, a majority
  45  * of matchIndex[i] ≥ N, and log[N].term == currentTerm:
  46  * set commitIndex = N (§5.3, §5.4).
  47  */
  48 public class Leader extends AbstractRaftActorBehavior {
  49
  50
  51     private final Map<String, ActorRef> followerToReplicator = new HashMap<>();
  52
  53     public Leader(RaftActorContext context, List<String> followers){
  54         super(context);
  55
  56         for(String follower : followers) {
  57
  58             ActorRef replicator = context.actorOf(
  59                 RaftReplicator.props(
  60                     new FollowerLogInformationImpl(follower,
  61                         new AtomicLong(0),
  62                         new AtomicLong(0)),
  63                     context.getActor()
  64                 )
  65             );
  66
  67             // Create a replicator for each follower
  68             followerToReplicator.put(follower, replicator);
  69
  70         }
  71
  72     }
  73
  74     @Override public RaftState handleMessage(ActorRef sender, Object message) {
  75         Preconditions.checkNotNull(sender, "sender should not be null");
  76
  77         if(message instanceof SendHeartBeat) {
  78             sender.tell(new AppendEntries(
  79                 context.getTermInformation().getCurrentTerm().get() , context.getId(),
  80                 context.getReplicatedLog().last().getIndex(),
  81                 context.getReplicatedLog().last().getTerm(),
  82                 Collections.EMPTY_LIST), context.getActor());
  83         }
  84         return RaftState.Leader;
  85     }
  86 }