Merge pull request #15398 from xakdwch/feature-enhance

rafthttp: replace inline code with existing function

server/etcdserver/api/rafthttp/pipeline.go

@@ -103,7 +103,7 @@ func (p *pipeline) handle() {
 			if err != nil {
 				p.status.deactivate(failureType{source: pipelineMsg, action: "write"}, err.Error())
 
-				if m.Type == raftpb.MsgApp && p.followerStats != nil {
+				if isMsgApp(m) && p.followerStats != nil {
 					p.followerStats.Fail()
 				}
 				p.raft.ReportUnreachable(m.To)
@@ -115,7 +115,7 @@ func (p *pipeline) handle() {
 			}
 
 			p.status.activate()
-			if m.Type == raftpb.MsgApp && p.followerStats != nil {
+			if isMsgApp(m) && p.followerStats != nil {
 				p.followerStats.Succ(end.Sub(start))
 			}
 			if isMsgSnap(m) {

server/etcdserver/api/rafthttp/transport.go

@@ -186,7 +186,7 @@ func (t *Transport) Send(msgs []raftpb.Message) {
 		t.mu.RUnlock()
 
 		if pok {
-			if m.Type == raftpb.MsgApp {
+			if isMsgApp(m) {
 				t.ServerStats.SendAppendReq(m.Size())
 			}
 			p.send(m)