func (r *Raft) VerifyLeader() Future {
	metrics.IncrCounter([]string{"raft", "verify_leader"}, 1)
	verifyFuture := &verifyFuture{}
	verifyFuture.init()
	select {
	case <-r.shutdownCh:
		return errorFuture{ErrRaftShutdown}
	case r.verifyCh <- verifyFuture:
		return verifyFuture
	}
}
VerifyLeader is used to ensure the current node is still
the leader. This can be done to prevent stale reads when a
new leader has potentially been elected.
package raft
import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"log"
	"net"
	"os"
	"strconv"
	"sync"
	"time"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/armon/go-metrics"
)
const (
	minCheckInterval = 10 * time.Millisecond
)
var (
	keyCurrentTerm  = []byte("CurrentTerm")
	keyLastVoteTerm = []byte("LastVoteTerm")
	keyLastVoteCand = []byte("LastVoteCand")
	ErrLeader = errors.New("node is the leader")
	ErrNotLeader = errors.New("node is not the leader")
	ErrLeadershipLost = errors.New("leadership lost while committing log")
	ErrRaftShutdown = errors.New("raft is already shutdown")
	ErrEnqueueTimeout = errors.New("timed out enqueuing operation")
	ErrKnownPeer = errors.New("peer already known")
	ErrUnknownPeer = errors.New("peer is unknown")
)
type commitTuple struct {
	log    *Log
	future *logFuture
}
type leaderState struct {
	commitCh  chan struct{}
	inflight  *inflight
	replState map[string]*followerReplication
	notify    map[*verifyFuture]struct{}
	stepDown  chan struct{}
}
type Raft struct {
	raftState
	applyCh chan *logFuture
	conf *Config
	fsm FSM
	fsmCommitCh chan commitTuple
	fsmRestoreCh chan *restoreFuture
	fsmSnapshotCh chan *reqSnapshotFuture
	lastContact     time.Time
	lastContactLock sync.RWMutex
	leader     net.Addr
	leaderLock sync.RWMutex
	leaderCh chan bool
	leaderState leaderState
	localAddr net.Addr
	logger *log.Logger
	logs LogStore
	peerCh    chan *peerFuture
	peers     []net.Addr
	peerStore PeerStore
	rpcCh <-chan RPC
	shutdown     bool
	shutdownCh   chan struct{}
	shutdownLock sync.Mutex
	snapshots SnapshotStore
	snapshotCh chan *snapshotFuture
	stable StableStore
	trans Transport
	verifyCh chan *verifyFuture
}
func NewRaft(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps SnapshotStore,
	peerStore PeerStore, trans Transport) (*Raft, error) {
	if err := ValidateConfig(conf); err != nil {
		return nil, err
	}
	if conf.LogOutput == nil {
		conf.LogOutput = os.Stderr
	}
	currentTerm, err := stable.GetUint64(keyCurrentTerm)
	if err != nil && err.Error() != "not found" {
		return nil, fmt.Errorf("failed to load current term: %v", err)
	}
	lastIdx, err := logs.LastIndex()
	if err != nil {
		return nil, fmt.Errorf("failed to find last log: %v", err)
	}
	var lastLog Log
	if lastIdx > 0 {
		if err := logs.GetLog(lastIdx, &lastLog); err != nil {
			return nil, fmt.Errorf("failed to get last log: %v", err)
		}
	}
	localAddr := trans.LocalAddr()
	peers, err := peerStore.Peers()
	if err != nil {
		return nil, fmt.Errorf("failed to get list of peers: %v", err)
	}
	peers = ExcludePeer(peers, localAddr)
	r := &Raft{
		applyCh:       make(chan *logFuture),
		conf:          conf,
		fsm:           fsm,
		fsmCommitCh:   make(chan commitTuple, 128),
		fsmRestoreCh:  make(chan *restoreFuture),
		fsmSnapshotCh: make(chan *reqSnapshotFuture),
		leaderCh:      make(chan bool),
		localAddr:     localAddr,
		logger:        log.New(conf.LogOutput, "", log.LstdFlags),
		logs:          logs,
		peerCh:        make(chan *peerFuture),
		peers:         peers,
		peerStore:     peerStore,
		rpcCh:         trans.Consumer(),
		snapshots:     snaps,
		snapshotCh:    make(chan *snapshotFuture),
		shutdownCh:    make(chan struct{}),
		stable:        stable,
		trans:         trans,
		verifyCh:      make(chan *verifyFuture, 64),
	}
	r.setState(Follower)
	r.setCurrentTerm(currentTerm)
	r.setLastLogIndex(lastLog.Index)
	r.setLastLogTerm(lastLog.Term)
	if err := r.restoreSnapshot(); err != nil {
		return nil, err
	}
	trans.SetHeartbeatHandler(r.processHeartbeat)
	r.goFunc(r.run)
	r.goFunc(r.runFSM)
	r.goFunc(r.runSnapshots)
	return r, nil
}
func (r *Raft) Leader() net.Addr {
	r.leaderLock.RLock()
	leader := r.leader
	r.leaderLock.RUnlock()
	return leader
}
func (r *Raft) setLeader(leader net.Addr) {
	r.leaderLock.Lock()
	r.leader = leader
	r.leaderLock.Unlock()
}
func (r *Raft) Apply(cmd []byte, timeout time.Duration) ApplyFuture {
	metrics.IncrCounter([]string{"raft", "apply"}, 1)
	var timer <-chan time.Time
	if timeout > 0 {
		timer = time.After(timeout)
	}
	logFuture := &logFuture{
		log: Log{
			Type: LogCommand,
			Data: cmd,
		},
	}
	logFuture.init()
	select {
	case <-timer:
		return errorFuture{ErrEnqueueTimeout}
	case <-r.shutdownCh:
		return errorFuture{ErrRaftShutdown}
	case r.applyCh <- logFuture:
		return logFuture
	}
}
func (r *Raft) Barrier(timeout time.Duration) Future {
	metrics.IncrCounter([]string{"raft", "barrier"}, 1)
	var timer <-chan time.Time
	if timeout > 0 {
		timer = time.After(timeout)
	}
	logFuture := &logFuture{
		log: Log{
			Type: LogBarrier,
		},
	}
	logFuture.init()
	select {
	case <-timer:
		return errorFuture{ErrEnqueueTimeout}
	case <-r.shutdownCh:
		return errorFuture{ErrRaftShutdown}
	case r.applyCh <- logFuture:
		return logFuture
	}
}