public interface CommandLog {
    long size();
    default long lastIndex() {
        return size() - 1;
    }
    int readTerm(long index);
    void readTo(long index, LogKey target);
    void readTo(long index, CommandKey target);
    void readTo(long index, LogEntry target);

    void append(int term, Command command);
    void truncateIncluding(long index);

    default int lastTerm() {
        return readTerm(lastIndex());
    }

    default void lastKeyTo(final LogKey target) {
        readTo(lastIndex(), target);
    }

    default int lastKeyCompareTo(final LogKey logKey) {
        final int termCompare = Integer.compare(lastTerm(), logKey.term());
        return termCompare == 0 ? Long.compare(lastIndex(), logKey.index()) : termCompare;
    }

    boolean contains(CommandKey commandKey);
    default LogContainment contains(final LogKey logKey) {
        return LogContainment.containmentFor(logKey, this);
    }

} public interface LogKey extends Comparable<LogKey> {
    Comparator<LogKey> COMPARATOR = (k1, k2) -> {
        final int termCompare = Integer.compare(k1.term(), k2.term());
        return termCompare == 0 ? Long.compare(k1.index(), k2.index()) : termCompare;
    };

    int term();
    LogKey term(int term);

    long index();
    LogKey index(long index);

    @Override
    default int compareTo(final LogKey other) {
        return COMPARATOR.compare(this, other);
    }

    default LogContainment containedIn(final CommandLog commandLog) {
        return LogContainment.containmentFor(this, commandLog);
    }

} public class AllocatingDirectFactory implements DirectFactory {

    private MutableDirectBuffer newBuffer(final int initialCapacity) {
        return new ExpandableArrayBuffer(initialCapacity);
    }

    @Override
    public AppendRequest appendRequest() {
        final DirectAppendRequest directAppendRequest = new DirectAppendRequest();
        directAppendRequest.wrap(newBuffer(DirectAppendRequest.EMPTY_LOG_BYTE_LENGTH), 0);
        return directAppendRequest;
    }

    @Override
    public AppendResponse appendResponse() {
        final DirectAppendResponse directAppendResponse = new DirectAppendResponse();
        directAppendResponse.wrap(newBuffer(DirectAppendResponse.BYTE_LENGTH), 0);
        return directAppendResponse;
    }

    @Override
    public VoteRequest voteRequest() {
        final DirectVoteRequest directVoteRequest = new DirectVoteRequest();
        directVoteRequest.wrap(newBuffer(DirectVoteRequest.BYTE_LENGTH), 0);
        return directVoteRequest;
    }

    @Override
    public VoteResponse voteResponse() {
        final DirectVoteResponse directVoteResponse = new DirectVoteResponse();
        directVoteResponse.wrap(newBuffer(DirectVoteResponse.BYTE_LENGTH), 0);
        return directVoteResponse;
    }

    @Override
    public TimeoutNow timeoutNow() {
        final DirectTimeoutNow directTimeoutNow = new DirectTimeoutNow();
        directTimeoutNow.wrap(newBuffer(DirectTimeoutNow.BYTE_LENGTH), 0);
        return null;
    }

    @Override
    public CommandKey commandKey() {
        final DirectCommandKey directCommandKey = new DirectCommandKey();
        directCommandKey.wrap(newBuffer(DirectCommandKey.BYTE_LENGTH), 0);
        return directCommandKey;
    }

    @Override
    public Command command() {
        final DirectCommand directCommand = new DirectCommand();
        directCommand.wrap(newBuffer(DirectCommand.EMPTY_COMMAND_BYTE_LENGTH), 0);
        return directCommand;
    }

    @Override
    public LogEntry logEntry() {
        final DirectLogEntry directCommandLogEntry = new DirectLogEntry();
        directCommandLogEntry.wrap(newBuffer(DirectLogEntry.EMPTY_COMMAND_BYTE_LENGTH), 0);
        return directCommandLogEntry;
    }
} public final class VoteRequestHandler {

    private final PersistentState persistentState;

    public VoteRequestHandler(final PersistentState persistentState) {
        this.persistentState = Objects.requireNonNull(persistentState);
    }

    public Transition onVoteRequest(final ServerContext serverContext, final VoteRequest voteRequest) {
        final CommandLog commandLog = persistentState.commandLog();
        final int term = voteRequest.term();
        final int candidateId = voteRequest.candidateId();
        final Transition transition;
        final boolean granted;
        if (persistentState.currentTerm() <= term && commandLog.lastKeyCompareTo(voteRequest.lastLogKey()) <= 0) {
            if (persistentState.votedFor() == PersistentState.NOT_VOTED_YET) {
                persistentState.votedFor(candidateId);
                //Why transition is TO_FOLLOWER?
                //in this case we should not replay the event!?
                transition = Transition.TO_FOLLOWER;
                granted = true;
            } else {
                granted = persistentState.votedFor() == candidateId;
                transition = Transition.STEADY;
            }
        } else {
            granted = false;
            transition = Transition.STEADY;
        }
        serverContext.directFactory().voteResponse()
                .term(persistentState.currentTerm())
                .voteGranted(granted)
                .sendTo(serverContext.connections().serverSender(candidateId),
                        serverContext.resendStrategy());
        return transition;
    }
} public interface Message extends DirectPayload, Event {

    MessageType type();

    default void sendTo(final Sender<? super Message> sender, final ResendStrategy resendStrategy) {
        final long res = sender.offer(this);
        if (res < 0) {
            resendStrategy.onRejectedOffer(sender, this, res);
        }
    }


} public interface VoteRequest extends Message {

    int term();

    VoteRequest term(int term);

    int candidateId();

    VoteRequest candidateId(int candidateId);

    LogKey lastLogKey();

    @Override
    default Transition accept(final ServerContext serverContext, final EventHandler eventHandler) {
        return eventHandler.onVoteRequest(serverContext, this);
    }
} public interface VoteResponse extends Message {

    int term();

    VoteResponse term(int term);

    boolean voteGranted();

    VoteResponse voteGranted(boolean granted);

    @Override
    default Transition accept(final ServerContext serverContext, final EventHandler eventHandler) {
        return eventHandler.onVoteResponse(serverContext, this);
    }
} public interface PersistentState {

    int NOT_VOTED_YET = -1;

    int currentTerm();

    int votedFor();

    int clearVotedForAndSetCurrentTerm(int term);

    int clearVotedForAndIncCurrentTerm();

    void votedFor(final int candidateId);

    CommandLog commandLog();
} public enum Role {
    CANDIDATE(CandidateState::new),
    LEADER(LeaderState::new),
    FOLLOWER(FollowerState::new);

    private final BiFunction<PersistentState, VolatileState, State> stateFactory;

    Role(final BiFunction<PersistentState, VolatileState, State> stateFactory) {
        this.stateFactory = Objects.requireNonNull(stateFactory);
    }

    public State createState(final PersistentState persistentState, final VolatileState volatileState) {
        return stateFactory.apply(persistentState, volatileState);
    }
} public interface Sender<M extends DirectPayload> {
    /**
     * Terminates composing and offsers the message to the transport in
     * non-blocking mode. Can be called multiple times for same message if sending
     * failed which is indicated through a negative return value.
     *
     * @param message the message to send
     * @return Non-negative transpport state such as position, otherwise zero if
     * not applicable and successful or a negative error value
     * {@link RejectReason#NOT_CONNECTED}, {@link RejectReason#BACK_PRESSURED},
     * {@link RejectReason#ADMIN_ACTION} or {@link RejectReason#CLOSED}.
     */
    long offer(M message);
}
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import org.tools4j.hoverraft.command.CommandLog;
import org.tools4j.hoverraft.command.LogKey;
import org.tools4j.hoverraft.direct.AllocatingDirectFactory;
import org.tools4j.hoverraft.event.VoteRequestHandler;
import org.tools4j.hoverraft.message.Message;
import org.tools4j.hoverraft.message.VoteRequest;
import org.tools4j.hoverraft.message.VoteResponse;
import org.tools4j.hoverraft.state.PersistentState;
import org.tools4j.hoverraft.state.Role;
import org.tools4j.hoverraft.transport.Sender;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
 * The MIT License (MIT)
 *
 * Copyright (c) 2016-2017 hover-raft (tools4j), Marco Terzer
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */
package org.tools4j.hoverraft.server;



@RunWith(MockitoJUnitRunner.class)
public class VoteRequestHandlerTest {

    private static final boolean GRANTED = true;
    private static final boolean REJECTED = false;

    //under test
    private VoteRequestHandler handler;