Method SendMsg

stream.go:939–1008

(m any)

}

func (cs *clientStream) SendMsg(m any) (err error) {
	defer func() {
		if err != nil && err != io.EOF {
			// Call finish on the client stream for errors generated by this SendMsg
			// call, as these indicate problems created by this client.  (Transport
			// errors are converted to an io.EOF error in csAttempt.sendMsg; the real
			// error will be returned from RecvMsg eventually in that case, or be
			// retried.)
			cs.finish(err)
		}
	}()
	if cs.sentLast {
		return status.Errorf(codes.Internal, "SendMsg called after CloseSend")
	}
	if !cs.desc.ClientStreams {
		cs.sentLast = true
	}

	// load hdr, payload, data
	hdr, data, payload, pf, err := prepareMsg(m, cs.codec, cs.compressorV0, cs.compressorV1, cs.cc.dopts.copts.BufferPool)
	if err != nil {
		return err
	}

	defer func() {
		data.Free()
		// only free payload if compression was made, and therefore it is a different set
		// of buffers from data.
		if pf.isCompressed() {
			payload.Free()
		}
	}()

	dataLen := data.Len()
	payloadLen := payload.Len()
	// TODO(dfawley): should we be checking len(data) instead?
	if payloadLen > *cs.callInfo.maxSendMessageSize {
		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", payloadLen, *cs.callInfo.maxSendMessageSize)
	}

	// always take an extra ref in case data == payload (i.e. when the data isn't
	// compressed). The original ref will always be freed by the deferred free above.
	payload.Ref()
	op := func(a *csAttempt) error {
		return a.sendMsg(m, hdr, payload, dataLen, payloadLen)
	}

	// onSuccess is invoked when the op is captured for a subsequent retry. If the
	// stream was established by a previous message and therefore retries are
	// disabled, onSuccess will not be invoked, and payloadRef can be freed
	// immediately.
	onSuccessCalled := false
	err = cs.withRetry(op, func() {
		cs.bufferForRetryLocked(len(hdr)+payloadLen, op, payload.Free)
		onSuccessCalled = true
	})
	if !onSuccessCalled {
		payload.Free()