func (r *Node) Apply() chan ApplyEntry {
	return r.applyc
}
Apply return the apply channel
package raft
import (
	"sync"
	"time"
	"sync/atomic"
	"encoding/binary"
	"github.com/coreos/etcd/pkg/contention"
	etcdraft "github.com/coreos/etcd/raft"
	"github.com/coreos/etcd/raft/raftpb"
	"github.com/coreos/etcd/rafthttp"
	"github.com/flike/kingbus/log"
	"github.com/flike/kingbus/storage"
	"github.com/flike/kingbus/utils"
)
const (
	maxInFlightMsgSnap = 16
)
type Node struct {
	tickMu *sync.Mutex
	NodeConfig
	msgSnapC chan raftpb.Message
	applyc       chan ApplyEntry
	appliedIndex uint64  
	committedIndex uint64  
	term           uint64  
	readStateC chan etcdraft.ReadState
	ticker *time.Ticker
	td *contention.TimeoutDetector
	stopped chan struct{}
	done    chan struct{}
}
type ApplyEntry struct {
	Entries  []raftpb.Entry
	Snapshot raftpb.Snapshot
	Notifyc chan struct{}
}
type NodeConfig struct {
	IsIDRemoved func(id uint64) bool
	etcdraft.Node
	Storage storage.Storage
	Heartbeat time.Duration  
	Transport rafthttp.Transporter
	PeerListener []*peerListener
}
func NewNode(cfg NodeConfig) *Node {
	r := &Node{
		tickMu:     new(sync.Mutex),
		NodeConfig: cfg,
		td:         contention.NewTimeoutDetector(2 * cfg.Heartbeat),
		readStateC: make(chan etcdraft.ReadState, 1),
		msgSnapC:   make(chan raftpb.Message, maxInFlightMsgSnap),
		applyc:     make(chan ApplyEntry),
		stopped:    make(chan struct{}),
		done:       make(chan struct{}),
	}
	if r.Heartbeat == 0 {
		r.ticker = &time.Ticker{}
	} else {
		r.ticker = time.NewTicker(r.Heartbeat)
	}
	return r
}
func (r *Node) tick() {
	r.tickMu.Lock()
	r.Tick()
	r.tickMu.Unlock()
}
func (r *Node) GetAppliedIndex() uint64 {
	return atomic.LoadUint64(&r.appliedIndex)
}
func (r *Node) SetAppliedIndex(v uint64) {
	atomic.StoreUint64(&r.appliedIndex, v)
}
func (r *Node) SetCommittedIndex(v uint64) {
	atomic.StoreUint64(&r.committedIndex, v)
}
func (r *Node) GetCommittedIndex() uint64 {
	return atomic.LoadUint64(&r.committedIndex)
}
func (r *Node) SetTerm(v uint64) {
	atomic.StoreUint64(&r.term, v)
}
func (r *Node) GetTerm() uint64 {
	return atomic.LoadUint64(&r.term)
}
func MustGetAppliedIndex(s storage.Storage) uint64 {
	value, err := s.Get(utils.StringToBytes(storage.AppliedIndexKey))
	if err != nil {
		log.Log.Fatalf("MustGetAppliedIndex:Get error,err:%s,key:%s", err.Error(), storage.AppliedIndexKey)
	}
	if len(value) == 0 {
		log.Log.Warningf("key:%s not exist,return 0", storage.AppliedIndexKey)
		return 0
	}
	return binary.BigEndian.Uint64(value)
}
type ReadyHandler struct {
	GetLead              func() (lead uint64)
	UpdateLead           func(lead uint64)
	UpdateLeadership     func(leadChange bool)
	UpdateCommittedIndex func(uint64)
}
func (r *Node) Stop() {
	select {
	case r.stopped <- struct{}{}:
	case <-r.done:
		return
	}
	<-r.done
}