/*
 * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */

package org.opendaylight.controller.cluster.raft.behaviors;

import akka.actor.ActorRef;
import com.google.common.base.Preconditions;
import org.opendaylight.controller.cluster.raft.FollowerLogInformationImpl;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
import org.opendaylight.controller.cluster.raft.RaftReplicator;
import org.opendaylight.controller.cluster.raft.RaftState;
import org.opendaylight.controller.cluster.raft.internal.messages.SendHeartBeat;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

/**
 * The behavior of a RaftActor when it is in the Leader state
 * <p>
 * Leaders:
 * <ul>
 * <li> Upon election: send initial empty AppendEntries RPCs
 * (heartbeat) to each server; repeat during idle periods to
 * prevent election timeouts (§5.2)
 * <li> If command received from client: append entry to local log,
 * respond after entry applied to state machine (§5.3)
 * <li> If last log index ≥ nextIndex for a follower: send
 * AppendEntries RPC with log entries starting at nextIndex
 * <ul>
 * <li> If successful: update nextIndex and matchIndex for
 * follower (§5.3)
 * <li> If AppendEntries fails because of log inconsistency:
 * decrement nextIndex and retry (§5.3)
 * </ul>
 * <li> If there exists an N such that N > commitIndex, a majority
 * of matchIndex[i] ≥ N, and log[N].term == currentTerm:
 * set commitIndex = N (§5.3, §5.4).
 */
public class Leader extends AbstractRaftActorBehavior {


    private final Map<String, ActorRef> followerToReplicator = new HashMap<>();

    public Leader(RaftActorContext context, List<String> followers){
        super(context);

        for(String follower : followers) {

            ActorRef replicator = context.actorOf(
                RaftReplicator.props(
                    new FollowerLogInformationImpl(follower,
                        new AtomicLong(0),
                        new AtomicLong(0)),
                    context.getActor()
                )
            );

            // Create a replicator for each follower
            followerToReplicator.put(follower, replicator);

        }

    }

    @Override public RaftState handleMessage(ActorRef sender, Object message) {
        Preconditions.checkNotNull(sender, "sender should not be null");

        if(message instanceof SendHeartBeat) {
            sender.tell(new AppendEntries(
                context.getTermInformation().getCurrentTerm().get() , context.getId(),
                context.getReplicatedLog().last().getIndex(),
                context.getReplicatedLog().last().getTerm(),
                Collections.EMPTY_LIST), context.getActor());
        }
        return RaftState.Leader;
    }
}