node.go 668 B

123456789101112131415161718192021222324252627282930313233343536373839
  1. package raft
  2. import "sync"
  3. type Interface interface {
  4. Step(m Message)
  5. }
  6. type Node struct {
  7. lk sync.Mutex
  8. sm *stateMachine
  9. }
  10. func New(k, addr int, next Interface) *Node {
  11. n := &Node{
  12. sm: newStateMachine(k, addr),
  13. }
  14. return n
  15. }
  16. // Propose asynchronously proposes data be applied to the underlying state machine.
  17. func (n *Node) Propose(data []byte) {
  18. m := Message{Type: msgProp, Data: data}
  19. n.Step(m)
  20. }
  21. func (n *Node) Step(m Message) {
  22. n.lk.Lock()
  23. defer n.lk.Unlock()
  24. n.sm.Step(m)
  25. }
  26. // Next advances the commit index and returns any new
  27. // commitable entries.
  28. func (n *Node) Next() []Entry {
  29. n.lk.Lock()
  30. defer n.lk.Unlock()
  31. return n.sm.nextEnts()
  32. }