Method doSendMessage

conn/node.go:425–522

(to uint64, msgCh chan []byte)

}

func (n *Node) doSendMessage(to uint64, msgCh chan []byte) error {
	addr, has := n.Peer(to)
	if !has {
		return errors.Errorf("Do not have address of peer %#x", to)
	}
	pool, err := GetPools().Get(addr)
	if err != nil {
		return err
	}

	c := pb.NewRaftClient(pool.Get())
	ctx, span := otel.Tracer("").Start(context.Background(),
		fmt.Sprintf("RaftMessage-%d-to-%d", n.Id, to))
	defer span.End()

	mc, err := c.RaftMessage(ctx)
	if err != nil {
		return err
	}

	var packets, lastPackets uint64
	slurp := func(batch *pb.RaftBatch) {
		for {
			if len(batch.Payload.Data) > messageBatchSoftLimit {
				return
			}
			select {
			case data := <-msgCh:
				batch.Payload.Data = append(batch.Payload.Data, data...)
				packets++
			default:
				return
			}
		}
	}

	ctx = mc.Context()

	fastTick := time.Tick(5 * time.Second)
	ticker := time.Tick(3 * time.Minute)

	for {
		select {
		case data := <-msgCh:
			batch := &pb.RaftBatch{
				Context: n.RaftContext,
				Payload: &api.Payload{Data: data},
			}
			slurp(batch) // Pick up more entries from msgCh, if present.
			span.AddEvent(fmt.Sprintf("[to: %x] [Packets: %d] Sending data of length: %d.",
				to, packets, len(batch.Payload.Data)))
			if packets%10000 == 0 {
				glog.V(2).Infof("[to: %x] [Packets: %d] Sending data of length: %d.",
					to, packets, len(batch.Payload.Data))
			}
			packets++
			if err := mc.Send(batch); err != nil {
				span.AddEvent(fmt.Sprintf("Error while mc.Send: %v", err))