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 interface DirectFactory {

    AppendRequest appendRequest();

    AppendResponse appendResponse();

    VoteRequest voteRequest();

    VoteResponse voteResponse();

    TimeoutNow timeoutNow();

    CommandKey commandKey();

    Command command();

    LogEntry logEntry();
} public interface AppendRequest extends Message {

    int term();

    AppendRequest term(int term);

    int leaderId();

    AppendRequest leaderId(int leaderId);

    LogKey prevLogKey();

    long leaderCommit();

    AppendRequest leaderCommit(long leaderCommit);

    Sequence<LogEntry> logEntries();

    default AppendRequest appendLogEntry(final LogEntry logEntry) {
        logEntries().append(logEntry);
        return this;
    }

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

    int term();

    AppendResponse term(int term);

    boolean successful();

    AppendResponse successful(boolean successful);

    AppendResponse serverId(int serverId);

    int serverId();

    AppendResponse matchLogIndex(long matchLogIndex);

    long matchLogIndex();

    @Override
    default Transition accept(final ServerContext serverContext, final EventHandler eventHandler) {
        return eventHandler.onAppendResponse(serverContext, this);
    }
} 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 class FollowerState extends AbstractState {

    public FollowerState(final PersistentState persistentState, final VolatileState volatileState) {
        super(Role.FOLLOWER, persistentState, volatileState);
    }

    @Override
    protected EventHandler eventHandler() {
        return new EventHandler() {
            @Override
            public Transition onVoteRequest(final ServerContext serverContext, final VoteRequest voteRequest) {
                return FollowerState.this.onVoteRequest(serverContext, voteRequest);
            }

            @Override
            public Transition onAppendRequest(final ServerContext serverContext, final AppendRequest appendRequest) {
                return FollowerState.this.onAppendRequest(serverContext, appendRequest);
            }

            @Override
            public Transition onTimerEvent(final ServerContext serverContext, final TimerEvent timerEvent) {
                return FollowerState.this.onTimerEvent(serverContext, timerEvent);
            }

            @Override
            public Transition onTimeoutNow(final ServerContext serverContext, final TimeoutNow timeoutNow) {
                return FollowerState.this.onTimeoutNow(serverContext, timeoutNow);
            }

        };
    }


    private Transition onTimerEvent(final ServerContext serverContext, final TimerEvent timerEvent) {
        return Transition.TO_CANDIDATE;
    }

    private Transition onTimeoutNow(final ServerContext serverContext, final TimeoutNow timeoutNow) {
        if (timeoutNow.term() == currentTerm() && timeoutNow.candidateId() == serverContext.id()) {
            return Transition.TO_CANDIDATE;
        }
        return Transition.STEADY;
    }


} 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 Transition implements Event {
    STEADY(null, false),
    TO_FOLLOWER(Role.FOLLOWER, true),
    TO_CANDIDATE(Role.CANDIDATE, false),
    TO_LEADER(Role.LEADER, false);

    private final Role targetRole;
    private final boolean replayEvent;

    Transition(final Role targetRole, final boolean replayEvent) {
        this.targetRole = targetRole;//nullable
        this.replayEvent = replayEvent;
    }

    public final Role targetRole(final Role currentRole) {
        return this == STEADY ? currentRole : targetRole;
    }

    public final boolean replayEvent() {
        return replayEvent;
    }

    @Override
    public Transition accept(final ServerContext serverContext, final EventHandler eventHandler) {
        return eventHandler.onTransition(serverContext, this);
    }

    public static Transition startWith(final ServerContext serverContext, final Event event, final EventHandler eventHandler) {
        return event.accept(serverContext, eventHandler);
    }
    public final Transition ifSteadyThen(final ServerContext serverContext, final Event event, final EventHandler eventHandler) {
        if (this == STEADY) {
            return event.accept(serverContext, eventHandler);
        }
        return this;
    }
} public final class VolatileState {

    private final TrackedFollowerState[] trackedFollowerStates;

    @Deprecated //REMOVE this is now storted as State in HoverRaftMachine
    private Role role = Role.FOLLOWER;
    private long commitIndex = -1;
    private long lastApplied = -1;

    public VolatileState(final int serverId, final ConsensusConfig consensusConfig) {
        this.trackedFollowerStates = initFollowerStates(serverId, consensusConfig);
    }

    public long commitIndex() {
        return commitIndex;
    }

    public VolatileState commitIndex(final long commitIndex) {
        this.commitIndex = commitIndex;
        return this;
    }

    public long lastApplied() {
        return lastApplied;
    }

    public VolatileState lastApplied(final long lastApplied) {
        this.lastApplied = lastApplied;
        return this;
    }

    public int followerCount() {
        return trackedFollowerStates.length;
    }

    public TrackedFollowerState followerState(int index) {
        return trackedFollowerStates[index];
    }

    public void resetFollowersState(final long nextIndex) {
        for (final TrackedFollowerState fs : trackedFollowerStates) {
            fs.nextIndex(nextIndex)
                    .resetMatchIndex();
        }
    }

    public TrackedFollowerState followerStateById(int id) {
        for (final TrackedFollowerState fs : trackedFollowerStates) {
            if (fs.serverId() == id) {
                return fs;
            }
        }
        throw new IllegalArgumentException("no follower state found for id " + id);
    }

    private static TrackedFollowerState[] initFollowerStates(final int serverId, final ConsensusConfig consensusConfig) {
        final TrackedFollowerState[] states = new TrackedFollowerState[consensusConfig.serverCount() - 1];
        int index = 0;
        for (int i = 0; i < consensusConfig.serverCount(); i++) {
            final int id = consensusConfig.serverConfig(i).id();
            if (id != serverId) {
                states[index] = new TrackedFollowerState(id);
                index++;
            }
        }
        return states;
    }
} 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.*;
import org.tools4j.hoverraft.direct.AllocatingDirectFactory;
import org.tools4j.hoverraft.direct.DirectFactory;
import org.tools4j.hoverraft.message.AppendRequest;
import org.tools4j.hoverraft.message.AppendResponse;
import org.tools4j.hoverraft.message.Message;
import org.tools4j.hoverraft.state.FollowerState;
import org.tools4j.hoverraft.state.PersistentState;
import org.tools4j.hoverraft.state.Transition;
import org.tools4j.hoverraft.state.VolatileState;
import org.tools4j.hoverraft.transport.Sender;
import static org.assertj.core.api.Assertions.assertThat;
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 FollowerStateTest {

    //under test
    private FollowerState followerState;

    private ServerContext serverContext;
    private PersistentState persistentState;
    private VolatileState volatileState;

    @Mock
    private Sender<Message> sender;