raft: refactor Node.Add/Remove

raft/cluster_test.go

@@ -92,9 +92,11 @@ func buildCluster(size int) (nt *network, nodes []*Node) {
 	}
 	nt = newNetwork(nis...)
 
-	Dictate(nodes[0]).Next()
+	lead := Dictate(nodes[0])
+	lead.Next()
 	for i := 1; i < size; i++ {
-		nt.send(nodes[0].newConfMessage(configAdd, &Config{NodeId: i}))
+		lead.Add(i)
+		nt.send(lead.Msgs()...)
 		for j := 0; j < i; j++ {
 			nodes[j].Next()
 		}

raft/node.go

@@ -44,25 +44,23 @@ func New(id int, heartbeat, election tick) *Node {
 
 func Dictate(n *Node) *Node {
 	n.Step(Message{Type: msgHup})
-	n.Step(n.newConfMessage(configAdd, &Config{NodeId: n.Id()}))
+	n.Add(n.Id())
 	return n
 }
 
 func (n *Node) Id() int { return n.sm.id }
 
 // Propose asynchronously proposes data be applied to the underlying state machine.
-func (n *Node) Propose(data []byte) {
-	m := Message{Type: msgProp, Entries: []Entry{{Data: data}}}
+func (n *Node) Propose(data []byte) { n.propose(normal, data) }
+
+func (n *Node) propose(t int, data []byte) {
+	m := Message{Type: msgProp, Entries: []Entry{{Type: t, Data: data}}}
 	n.Step(m)
 }
 
-func (n *Node) Add(id int) {
-	n.Step(n.newConfMessage(configAdd, &Config{NodeId: id}))
-}
+func (n *Node) Add(id int) { n.updateConf(configAdd, &Config{NodeId: id}) }
 
-func (n *Node) Remove(id int) {
-	n.Step(n.newConfMessage(configRemove, &Config{NodeId: id}))
-}
+func (n *Node) Remove(id int) { n.updateConf(configRemove, &Config{NodeId: id}) }
 
 func (n *Node) Msgs() []Message {
 	return n.sm.Msgs()
@@ -132,10 +130,10 @@ func (n *Node) Tick() {
 	}
 }
 
-func (n *Node) newConfMessage(t int, c *Config) Message {
+func (n *Node) updateConf(t int, c *Config) {
 	data, err := json.Marshal(c)
 	if err != nil {
 		panic(err)
 	}
-	return Message{Type: msgProp, Entries: []Entry{Entry{Type: t, Data: data}}}
+	n.propose(t, data)
 }