Method SendMsg

stream.go:1711–1792

(m any)

}

func (ss *serverStream) SendMsg(m any) (err error) {
	defer func() {
		if ss.trInfo != nil {
			ss.mu.Lock()
			if ss.trInfo.tr != nil {
				if err == nil {
					ss.trInfo.tr.LazyLog(&payload{sent: true, msg: m}, true)
				} else {
					ss.trInfo.tr.LazyLog(&fmtStringer{"%v", []any{err}}, true)
					ss.trInfo.tr.SetError()
				}
			}
			ss.mu.Unlock()
		}
		if err != nil && err != io.EOF {
			st, _ := status.FromError(toRPCErr(err))
			ss.s.WriteStatus(st)
			// Non-user specified status was sent out. This should be an error
			// case (as a server side Cancel maybe).
			//
			// This is not handled specifically now. User will return a final
			// status from the service handler, we will log that error instead.
			// This behavior is similar to an interceptor.
		}
	}()

	// Server handler could have set new compressor by calling SetSendCompressor.
	// In case it is set, we need to use it for compressing outbound message.
	if sendCompressorsName := ss.s.SendCompress(); sendCompressorsName != ss.sendCompressorName {
		ss.compressorV1 = encoding.GetCompressor(sendCompressorsName)
		ss.sendCompressorName = sendCompressorsName
	}

	// load hdr, payload, data
	hdr, data, payload, pf, err := prepareMsg(m, ss.codec, ss.compressorV0, ss.compressorV1, ss.p.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 > ss.maxSendMessageSize {
		return status.Errorf(codes.ResourceExhausted, "trying to send message larger than max (%d vs. %d)", payloadLen, ss.maxSendMessageSize)
	}
	if err := ss.s.Write(hdr, payload, &transport.WriteOptions{Last: false}); err != nil {
		return toRPCErr(err)
	}