func (peer *Peer) SendMessage(cmnd Command) bool {
	select {
	case peer.sendMessage <- cmnd:
		return true
	default:
		return false
	}
}
SendMessage attempts to send a message to this peer.
A RequestVote is sent if this node is in Candidate state
An AppendEntries is sent if this node is in Leader state.
True is returned if message delivery can be attempted, false otherwise.
package server
import (
	"github.com/owlfish/forestbus-server/commitlog"
	"github.com/owlfish/forestbus-server/utils"
	"math"
	"sort"
	"sync"
	"time"
)
type NodeState int
func (state NodeState) String() string {
	switch state {
	case STARTING_UP_NODE:
		return "Starting Up"
	case FOLLOWER_NODE:
		return "Follower"
	case CANDIDATE_NODE:
		return "Candidate"
	case LEADER_NODE:
		return "Leader"
	case ORPHAN_NODE:
		return "Orphan"
	case SHUTDOWN_NODE:
		return "Shutdown"
	}
	return "Unknown"
}
const RAFT_NODE_SUBSYSTEM_SHUTDOWN_TIMEOUT = 500 * time.Millisecond
const (
	STARTING_UP_NODE NodeState = iota
	FOLLOWER_NODE
	CANDIDATE_NODE
	LEADER_NODE
	ORPHAN_NODE
	SHUTDOWN_NODE
)
type CommitIndex struct {
	firstIndexOfTerm   int64
	currentCommitIndex int64
	lock               sync.RWMutex
	waitingClients     *sync.Cond
}
func NewCommitIndex() *CommitIndex {
	c := &CommitIndex{}
	c.waitingClients = sync.NewCond(c.lock.RLocker())
	return c
}
func (commInd *CommitIndex) FirstIndexOfTerm(firstInd int64, lastSeenCommit int64) {
	commInd.lock.Lock()
	defer commInd.lock.Unlock()
	commInd.firstIndexOfTerm = firstInd
	if commInd.currentCommitIndex < lastSeenCommit {
		commInd.currentCommitIndex = lastSeenCommit
	}
}
func (commInd *CommitIndex) UpdateCommitIndex(node *Node) {
	commInd.lock.Lock()
	defer commInd.lock.Unlock()
	node.lock.RLock()
	defer node.lock.RUnlock()
	indexes := make([]float64, len(node.peers)+1)
	i := 0
	for _, peer := range node.peers {
		indexes[i] = float64(peer.GetLastIndex())
		i++
	}
	ourIndex, err := node.log.LastIndex()
	ourIndexFloat := float64(ourIndex)
	indexes[len(indexes)-1] = ourIndexFloat
	if err != nil {
		node.server.RequestShutdown("Error reading commit index from log file.")
	}
	sort.Float64s(indexes)
	cutOffIndex := int(math.Floor((float64(len(indexes)) / 2)))
	index := int64(indexes[cutOffIndex])
	if index >= commInd.firstIndexOfTerm {
		commInd.currentCommitIndex = index
		commInd.waitingClients.Broadcast()
	} else if commInd.currentCommitIndex == 0 {
		superMajority := true
		for _, peer := range node.peers {
			if peer.GetLastValidatedIndex() != ourIndex {
				superMajority = false
				break
			}
		}
		if superMajority {
			node.node_log("All of the peers agree on index, updating commit index.")
			commInd.currentCommitIndex = index
			commInd.waitingClients.Broadcast()
		}
	}
}
func (commInd *CommitIndex) GetCommitIndex() int64 {
	commInd.lock.RLock()
	defer commInd.lock.RUnlock()
	return commInd.currentCommitIndex
}
func (commInd *CommitIndex) WaitOnCommitChange(index int64) int64 {
	commInd.lock.RLock()
	defer commInd.lock.RUnlock()
	if index <= commInd.currentCommitIndex {
		return commInd.currentCommitIndex
	}
	commInd.waitingClients.Wait()
	return commInd.currentCommitIndex
}
func (commInd *CommitIndex) ReleaseWaitingClients() {
	commInd.lock.RLock()
	defer commInd.lock.RUnlock()
	commInd.waitingClients.Broadcast()
}
const MAX_INCOMING_MESSAGE_SIZE = 5 * 1024 * 1024
const LEADER_ACK_TIMEOUT = 200 * time.Millisecond
const CANDIDATE_VOTE_TIMEOUT = 200 * time.Millisecond
type TopicPersistentStore interface {
	SetTerm(term int64, votedFor string) (err error)
	Load() (term int64, votedFor string, err error)
}
type Command struct {
	Action        int
	ResultChannel chan interface{}
}
const (
	ACTION_SEND_VOTE int = iota
	ACTION_SEND_LOG
	ACTION_SEND_ACK
	ACTION_SHUTDOWN
)
type Peer struct {
	Name        string
	sendMessage chan Command
	lock        sync.RWMutex
	nextIndex   int64
	lastIndex int64
	validatedIndex int64
}
type PeerInfo struct {
	Name      string
	NextIndex int64
	LastIndex int64
}