Repository: grpc/grpc-go
File: stream.go (Go)

Code before:
			err = status.Error(codes.Internal, io.ErrUnexpectedEOF.Error())
		}
		return toRPCErr(err)
	}
	if len(ss.statsHandler) != 0 {
		for _, sh := range ss.statsHandler {
			sh.HandleRPC(ss.s.Context(), &stats.InPayload{
				RecvTime:         time.Now(),
				Payload:          m,
				Length:           payInfo.uncompressedBytes.Len(),
				WireLength:       payInfo.compressedLength + headerLen,
				CompressedLength: payInfo.compressedLength,
			})
		}
	}
	if len(ss.binlogs) != 0 {
		cm := &binarylog.ClientMessage{
			Message: payInfo.uncompressedBytes.Materialize(),
		}
		for _, binlog := range ss.binlogs {
			binlog.Log(ss.ctx, cm)
		}
	}
	// Special handling for non-client-stream rpcs.
	if !ss.desc.ClientStreams {
		if err := recv(ss.p, ss.codec, ss.s, ss.decompressorV0, m, ss.maxReceiveMessageSize, payInfo, ss.decompressorV1, true); err != io.EOF {
			return status.Errorf(codes.Internal, "cardinality violation: expected <EOF> for non client-streaming RPCs, but received another message")
		}
	}
	return nil
}

// MethodFromServerStream returns the method string for the input stream.
// The returned string is in the format of "/service/method".
func MethodFromServerStream(stream ServerStream) (string, bool) {
	return Method(stream.Context())
}

// prepareMsg returns the hdr, payload and data using the compressors passed or
// using the passed preparedmsg. The returned boolean indicates whether
// compression was made and therefore whether the payload needs to be freed in
// addition to the returned data. Freeing the payload if the returned boolean is
// false can lead to undefined behavior.
func prepareMsg(m any, codec baseCodec, cp Compressor, comp encoding.Compressor, pool mem.BufferPool) (hdr []byte, data, payload mem.BufferSlice, pf payloadFormat, err error) {
	if preparedMsg, ok := m.(*PreparedMsg); ok {
		return preparedMsg.hdr, preparedMsg.encodedData, preparedMsg.payload, preparedMsg.pf, nil
	}
	// The input interface is not a prepared msg.
	// Marshal and Compress the data at this point
	data, err = encode(codec, m)
	if err != nil {

Review comment (suggestion):
Similar to the client stream check, pass `nil` instead of `payInfo` and add a comment:
> This recv expects EOF or errors, so we don't collect inPayload. 

Code after:
			return err
		}
		if err == io.ErrUnexpectedEOF {
			err = status.Error(codes.Internal, io.ErrUnexpectedEOF.Error())
		}
		return toRPCErr(err)
	}
	ss.recvFirstMsg = true
	if len(ss.statsHandler) != 0 {
		for _, sh := range ss.statsHandler {
			sh.HandleRPC(ss.s.Context(), &stats.InPayload{
				RecvTime:         time.Now(),
				Payload:          m,
				Length:           payInfo.uncompressedBytes.Len(),
				WireLength:       payInfo.compressedLength + headerLen,
				CompressedLength: payInfo.compressedLength,
			})
		}
	}
	if len(ss.binlogs) != 0 {
		cm := &binarylog.ClientMessage{
			Message: payInfo.uncompressedBytes.Materialize(),
		}
		for _, binlog := range ss.binlogs {
			binlog.Log(ss.ctx, cm)
		}
	}

	if ss.desc.ClientStreams {
		// Subsequent messages should be received by subsequent RecvMsg calls.
		return nil
	}
	// Special handling for non-client-stream rpcs.
	// This recv expects EOF or errors, so we don't collect inPayload.
	if err := recv(ss.p, ss.codec, ss.s, ss.decompressorV0, m, ss.maxReceiveMessageSize, nil, ss.decompressorV1, true); err == io.EOF {
		return nil
	} else if err != nil {
		return err
	}
	return status.Error(codes.Internal, "cardinality violation: received multiple request messages for non-client-streaming RPC")
}

// MethodFromServerStream returns the method string for the input stream.
// The returned string is in the format of "/service/method".
func MethodFromServerStream(stream ServerStream) (string, bool) {
	return Method(stream.Context())
}

// prepareMsg returns the hdr, payload and data using the compressors passed or
// using the passed preparedmsg. The returned boolean indicates whether
// compression was made and therefore whether the payload needs to be freed in