Method RaftMessage

conn/raft_server.go:182–264

RaftMessage handles RAFT messages.

(server pb.Raft_RaftMessageServer)

// RaftMessage handles RAFT messages.
func (w *RaftServer) RaftMessage(server pb.Raft_RaftMessageServer) error {
	ctx := server.Context()
	if ctx.Err() != nil {
		return ctx.Err()
	}
	span := trace.SpanFromContext(ctx)

	node := w.GetNode()
	if node == nil || node.Raft() == nil {
		return ErrNoNode
	}
	span.AddEvent(fmt.Sprintf("Stream server is node %d", node.Id))

	var rc *pb.RaftContext
	raft := node.Raft()
	step := func(data []byte) error {
		ctx, cancel := context.WithTimeout(ctx, time.Minute)
		defer cancel()

		for idx := 0; idx < len(data); {
			x.AssertTruef(len(data[idx:]) >= 4,
				"Slice left of size: %v. Expected at least 4.", len(data[idx:]))

			sz := int(binary.LittleEndian.Uint32(data[idx : idx+4]))
			idx += 4
			msg := raftpb.Message{}
			if idx+sz > len(data) {
				return errors.Errorf(
					"Invalid query. Specified size %v overflows slice [%v,%v)\n",
					sz, idx, len(data))
			}
			if err := msg.Unmarshal(data[idx : idx+sz]); err != nil {
				x.Check(err)
			}
			// This should be done in order, and not via a goroutine.
			// Step can block forever. See: https://github.com/etcd-io/etcd/issues/10585
			// So, add a context with timeout to allow it to get out of the blockage.
			if glog.V(2) {
				switch msg.Type {
				case raftpb.MsgHeartbeat, raftpb.MsgHeartbeatResp:
					atomic.AddInt64(&node.heartbeatsIn, 1)
				case raftpb.MsgReadIndex, raftpb.MsgReadIndexResp:
				case raftpb.MsgApp, raftpb.MsgAppResp:
				case raftpb.MsgProp:
				default:
					glog.Infof("RaftComm: [%#x] Received msg of type: %s from %#x",
						msg.To, msg.Type, msg.From)
				}
			}
			if err := raft.Step(ctx, msg); err != nil {
				glog.Warningf("Error while raft.Step from %#x: %v. Closing RaftMessage stream.",
					rc.GetId(), err)
				return errors.Wrapf(err, "error while raft.Step from %#x", rc.GetId())
			}
			idx += sz
		}
		return nil
	}