public interface Handler<E> {

	public void handleEvent(E o);

	public String getName();
	
}
 public class QueueTask<E> implements Runnable {

	private Logger logger;
	
	private Queue<E> queue = new LinkedBlockingQueue<>();
	
	private AtomicBoolean running = new AtomicBoolean(false);
	
	private Lock lock = new ReentrantLock();
	
	private Handler<E> handler;
	
	public QueueTask(Handler<E> handler) {
		super();
		this.handler = handler;
		logger = LoggerFactory.getLogger(handler.getClass());
	}


	@Override
	public void run() {
		String oldName = Thread.currentThread().getName();
		Thread.currentThread().setName("" + handler.getName());
		while (true) {
			E o = queue.poll();
			if (o == null) {
				lock.lock();
				try {
					if (queue.isEmpty()) {
						running.set(false);
						Thread.currentThread().setName(oldName);
						return;
					} else {
						continue;
					}
				} finally {
					lock.unlock();
				}
			}
			try {
				handler.handleEvent(o);
			} catch (Exception e) {
				logger.error("Failed to handle: " + o + ": " + e, e);
			}			
		}
	}


	public AtomicBoolean getRunning() {
		return running;
	}

	public Lock getLock() {
		return lock;
	}


	public Queue<E> getQueue() {
		return queue;
	}
	
}
 public class AppendRequest {

	private long index;
	private long leaderTerm;
	private long logTerm;
	private long previousIndex;
	private long previousTerm;
	private long commitIndex;
	private Object payload;
	private transient MessageChannel clientChannel;
	
	public AppendRequest() {
		super();
	}

	public AppendRequest(long logTerm, long index, long leaderTerm, long previousIndex,
			long previousTerm, long commitIndex, Object payload) {
		super();
		this.logTerm = logTerm;
		this.index = index;
		this.leaderTerm = leaderTerm;
		this.previousIndex = previousIndex;
		this.previousTerm = previousTerm;
		this.commitIndex = commitIndex;
		this.payload = payload;
	}

	public AppendRequest(long logTerm, long index, long leaderTerm, long previousIndex,
			long previousTerm, long commitIndex, Object payload,
			MessageChannel clientChannel) {
		super();
		this.logTerm = logTerm;
		this.index = index;
		this.leaderTerm = leaderTerm;
		this.previousIndex = previousIndex;
		this.previousTerm = previousTerm;
		this.commitIndex = commitIndex;
		this.payload = payload;
		this.clientChannel = clientChannel;
	}

	public long getIndex() {
		return index;
	}

	public long getLeaderTerm() {
		return leaderTerm;
	}

	public long getLogTerm() {
		return logTerm;
	}

	public long getPreviousIndex() {
		return previousIndex;
	}

	public long getPreviousTerm() {
		return previousTerm;
	}

	public long getCommitIndex() {
		return commitIndex;
	}

	public Object getPayload() {
		return payload;
	}

	@JsonIgnore
	public MessageChannel getClientChannel() {
		return clientChannel;
	}

	@Override
	public String toString() {
		return "AppendRequest [index=" + index + ", term=" + leaderTerm
				+ ", previousIndex=" + previousIndex + ", previousTerm="
				+ previousTerm + ", commitIndex=" + commitIndex + ", "
				+ (payload != null ? "payload=" + payload : "") + "]";
	}

	
}
 public class AppendResponse {
	
	private long currentTerm;
	
	private boolean success;

	private long entryIndex;
	
	private long entryTerm; 
	
	
	public AppendResponse(long currentTerm, boolean success, long entryIndex,
			long entryTerm) {
		super();
		this.currentTerm = currentTerm;
		this.success = success;
		this.entryIndex = entryIndex;
		this.entryTerm = entryTerm;
	}

	public AppendResponse() {
		super();
	}

	public long getCurrentTerm() {
		return currentTerm;
	}

	public void setCurrentTerm(long term) {
		this.currentTerm = term;
	}

	public boolean isSuccess() {
		return success;
	}

	public void setSuccess(boolean success) {
		this.success = success;
	}

	public long getEntryIndex() {
		return entryIndex;
	}

	public long getEntryTerm() {
		return entryTerm;
	}

	@Override
	public String toString() {
		return "AppendResponse [currentTerm=" + currentTerm + ", success="
				+ success + ", entryIndex=" + entryIndex + ", entryTerm="
				+ entryTerm + "]";
	}
	

	

}
 public class ClientRequest {

	/**
	 * Payload of object, including reference to client MessageChannel
	 */
	private Object payload;

	
	public ClientRequest() {
		super();
	}

	public ClientRequest(Object payload) {
		super();
		this.payload = payload;
	}

	public Object getPayload() {
		return payload;
	}
	
	
}
 public class ClientResponse {

	private int errCode;
	
	private Object payload;

	public ClientResponse(int errCode, Object payload) {
		super();
		this.errCode = errCode;
		this.payload = payload;
	}

	public ClientResponse() {
		super();
	}

	public int getErrCode() {
		return errCode;
	}

	public Object getPayload() {
		return payload;
	}

	@Override
	public String toString() {
		return "ClientResponse [errCode=" + errCode + ", payload=" + payload
				+ "]";
	}
	
}
 public interface MessageChannel {

	public void send(Raft source, Object o);
	
}
 public class RaftEvent {

	public enum EventType {
		APPEND,
		APPEND_RESPONSE,
		REQUEST_VOTE,
		VOTE_GRANTED,
		LOOP, 
		CLIENT_REQUEST,
		;
	}
	
	private ClusterMember source;
	
	private MessageChannel clientSource;
	
	private EventType type;
	
	private Object event;

	public RaftEvent(ClusterMember source, EventType type, Object event) {
		super();
		this.source = source;
		this.type = type;
		this.event = event;
	}

	public RaftEvent(MessageChannel clientSource, EventType type, Object event) {
		super();
		this.clientSource = clientSource;
		this.type = type;
		this.event = event;
	}



	public ClusterMember getSource() {
		return source;
	}

	public EventType getType() {
		return type;
	}

	public Object getEvent() {
		return event;
	}

	public MessageChannel getClientSource() {
		return clientSource;
	}

	@Override
	public String toString() {
		return "RaftEvent ["
				+ (source != null ? "source=" + source + ", " : "")
				+ (clientSource != null ? "clientSource=" + clientSource + ", "
						: "") + (type != null ? "type=" + type + ", " : "")
				+ (event != null ? "event=" + event : "") + "]";
	}


	
}
 public class RequestForVote {

	private long term;
	
	private long lastLogIndex;
	
	private long lastLogTerm;
	
	public RequestForVote(long term, long lastLogIndex, long lastLogTerm) {
		super();
		this.term = term;
		this.lastLogIndex = lastLogIndex;
		this.lastLogTerm = lastLogTerm;
	}

	public RequestForVote() {
		super();
	}

	public long getTerm() {
		return term;
	}

	public void setTerm(long term) {
		this.term = term;
	}

	public long getLastLogIndex() {
		return lastLogIndex;
	}

	public void setLastLogIndex(long lastLogIndex) {
		this.lastLogIndex = lastLogIndex;
	}

	public long getLastLogTerm() {
		return lastLogTerm;
	}

	public void setLastLogTerm(long lastLogTerm) {
		this.lastLogTerm = lastLogTerm;
	}

	@Override
	public String toString() {
		return "RequestForVote [term=" + term + ", lastLogIndex="
				+ lastLogIndex + ", lastLogTerm=" + lastLogTerm + "]";
	}

}
 public class VoteGranted {

	private long term;
	
	public VoteGranted() {
		super();
	}
	
	public VoteGranted(long term) {
		super();
		this.term = term;
	}

	public long getTerm() {
		return term;
	}
	
}
 public enum EventType {
	APPEND,
	APPEND_RESPONSE,
	REQUEST_VOTE,
	VOTE_GRANTED,
	LOOP, 
	CLIENT_REQUEST,
	;
}

import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.r4j.executor.Handler;
import org.r4j.executor.QueueTask;
import org.r4j.protocol.AppendRequest;
import org.r4j.protocol.AppendResponse;
import org.r4j.protocol.ClientRequest;
import org.r4j.protocol.ClientResponse;
import org.r4j.protocol.MessageChannel;
import org.r4j.protocol.RaftEvent;
import org.r4j.protocol.RequestForVote;
import org.r4j.protocol.VoteGranted;
import org.r4j.protocol.RaftEvent.EventType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

package org.r4j;



public class RaftEventQueue implements RaftListener, Handler<RaftEvent> {

	private Logger logger = LoggerFactory.getLogger(getClass());
	
	private Raft raft;

	private ScheduledExecutorService executor;

	private QueueTask<RaftEvent> queue = new QueueTask<>(this);

	private LoopEvent loopEvent = new LoopEvent();
	
	private boolean running = true;

	public RaftEventQueue(Raft raft, ScheduledExecutorService executor) {
		super();
		this.raft = raft;
		this.executor = executor;
	}

	public void setMembers(List<ClusterMember> remoteMembers) {
		raft.setMembers(remoteMembers);
	}

	public void init() {
		running = true;
		loop(1L);
	}

	private void loop(long l) {
		if (running) {
			if (loopEvent.scheduleCounter.get() < 1) {
				loopEvent.scheduleCounter.incrementAndGet();
				executor.schedule(loopEvent, l, TimeUnit.MILLISECONDS);				
			} else {
				logger.error("did not schedule loop for: " + raft);
			}
		}
	}

	@Override
	public void handleEvent(RaftEvent e) {
		if (!running) {
			if (e.getType() == EventType.CLIENT_REQUEST) {
				e.getClientSource().send(raft, new ClientResponse(301, null));
			}
			return;
		}
		switch (e.getType()) {
		case APPEND:
			AppendRequest ar = (AppendRequest) e.getEvent();
			boolean appended = raft.append(ar);
			if (ar.getPayload() != null) {
				logger.info("appended: " + appended + " i=" + ar.getIndex() + " t=" + ar.getLogTerm() + " " + raft);
				e.getSource().getChannel()
					.send(raft, new AppendResponse(raft.getTerm().getCurrent(), appended, ar.getIndex(), ar.getLogTerm()));
			} else {
				logger.info("ping received: i=" + ar.getIndex() + " t=" + ar.getLeaderTerm() + " " + raft);
			}
			break;
		case APPEND_RESPONSE:
			raft.handleResponse(e.getSource(), (AppendResponse)e.getEvent());
			logger.info("append responded: " + e.getEvent() + " " + raft);
			break;
		case REQUEST_VOTE:
			RequestForVote rfv = (RequestForVote) e.getEvent();
			if (raft.getElection().vote(e.getSource(), rfv)) {
				logger.info("Voted for: " + rfv);

				e.getSource().getChannel().send(raft, new VoteGranted(rfv.getTerm()));
