2019-11-07 20:05:39 +01:00
/ *
*
* Copyright 2014 gRPC authors .
*
* Licensed under the Apache License , Version 2.0 ( the "License" ) ;
* you may not use this file except in compliance with the License .
* You may obtain a copy of the License at
*
* http : //www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing , software
* distributed under the License is distributed on an "AS IS" BASIS ,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND , either express or implied .
* See the License for the specific language governing permissions and
* limitations under the License .
*
* /
package grpc
import (
"bytes"
"compress/gzip"
"context"
"encoding/binary"
"fmt"
"io"
"math"
"strings"
"sync"
"time"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/encoding"
"google.golang.org/grpc/encoding/proto"
"google.golang.org/grpc/internal/transport"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/peer"
"google.golang.org/grpc/stats"
"google.golang.org/grpc/status"
)
// Compressor defines the interface gRPC uses to compress a message.
//
// Deprecated: use package encoding.
type Compressor interface {
// Do compresses p into w.
Do ( w io . Writer , p [ ] byte ) error
// Type returns the compression algorithm the Compressor uses.
Type ( ) string
}
type gzipCompressor struct {
pool sync . Pool
}
// NewGZIPCompressor creates a Compressor based on GZIP.
//
// Deprecated: use package encoding/gzip.
func NewGZIPCompressor ( ) Compressor {
c , _ := NewGZIPCompressorWithLevel ( gzip . DefaultCompression )
return c
}
// NewGZIPCompressorWithLevel is like NewGZIPCompressor but specifies the gzip compression level instead
// of assuming DefaultCompression.
//
// The error returned will be nil if the level is valid.
//
// Deprecated: use package encoding/gzip.
func NewGZIPCompressorWithLevel ( level int ) ( Compressor , error ) {
if level < gzip . DefaultCompression || level > gzip . BestCompression {
return nil , fmt . Errorf ( "grpc: invalid compression level: %d" , level )
}
return & gzipCompressor {
pool : sync . Pool {
2023-09-07 12:34:14 +02:00
New : func ( ) any {
2023-02-08 17:55:14 +01:00
w , err := gzip . NewWriterLevel ( io . Discard , level )
2019-11-07 20:05:39 +01:00
if err != nil {
panic ( err )
}
return w
} ,
} ,
} , nil
}
func ( c * gzipCompressor ) Do ( w io . Writer , p [ ] byte ) error {
z := c . pool . Get ( ) . ( * gzip . Writer )
defer c . pool . Put ( z )
z . Reset ( w )
if _ , err := z . Write ( p ) ; err != nil {
return err
}
return z . Close ( )
}
func ( c * gzipCompressor ) Type ( ) string {
return "gzip"
}
// Decompressor defines the interface gRPC uses to decompress a message.
//
// Deprecated: use package encoding.
type Decompressor interface {
// Do reads the data from r and uncompress them.
Do ( r io . Reader ) ( [ ] byte , error )
// Type returns the compression algorithm the Decompressor uses.
Type ( ) string
}
type gzipDecompressor struct {
pool sync . Pool
}
// NewGZIPDecompressor creates a Decompressor based on GZIP.
//
// Deprecated: use package encoding/gzip.
func NewGZIPDecompressor ( ) Decompressor {
return & gzipDecompressor { }
}
func ( d * gzipDecompressor ) Do ( r io . Reader ) ( [ ] byte , error ) {
var z * gzip . Reader
switch maybeZ := d . pool . Get ( ) . ( type ) {
case nil :
newZ , err := gzip . NewReader ( r )
if err != nil {
return nil , err
}
z = newZ
case * gzip . Reader :
z = maybeZ
if err := z . Reset ( r ) ; err != nil {
d . pool . Put ( z )
return nil , err
}
}
defer func ( ) {
z . Close ( )
d . pool . Put ( z )
} ( )
2023-02-08 17:55:14 +01:00
return io . ReadAll ( z )
2019-11-07 20:05:39 +01:00
}
func ( d * gzipDecompressor ) Type ( ) string {
return "gzip"
}
// callInfo contains all related configuration and information about an RPC.
type callInfo struct {
compressorType string
failFast bool
maxReceiveMessageSize * int
maxSendMessageSize * int
creds credentials . PerRPCCredentials
contentSubtype string
codec baseCodec
maxRetryRPCBufferSize int
2023-03-25 02:08:06 +01:00
onFinish [ ] func ( err error )
2019-11-07 20:05:39 +01:00
}
func defaultCallInfo ( ) * callInfo {
return & callInfo {
failFast : true ,
maxRetryRPCBufferSize : 256 * 1024 , // 256KB
}
}
// CallOption configures a Call before it starts or extracts information from
// a Call after it completes.
type CallOption interface {
// before is called before the call is sent to any server. If before
// returns a non-nil error, the RPC fails with that error.
before ( * callInfo ) error
// after is called after the call has completed. after cannot return an
// error, so any failures should be reported via output parameters.
2020-06-25 22:42:41 +02:00
after ( * callInfo , * csAttempt )
2019-11-07 20:05:39 +01:00
}
// EmptyCallOption does not alter the Call configuration.
// It can be embedded in another structure to carry satellite data for use
// by interceptors.
type EmptyCallOption struct { }
2020-06-25 22:42:41 +02:00
func ( EmptyCallOption ) before ( * callInfo ) error { return nil }
func ( EmptyCallOption ) after ( * callInfo , * csAttempt ) { }
2019-11-07 20:05:39 +01:00
// Header returns a CallOptions that retrieves the header metadata
// for a unary RPC.
func Header ( md * metadata . MD ) CallOption {
return HeaderCallOption { HeaderAddr : md }
}
// HeaderCallOption is a CallOption for collecting response header metadata.
// The metadata field will be populated *after* the RPC completes.
2020-10-16 14:04:44 +02:00
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type HeaderCallOption struct {
HeaderAddr * metadata . MD
}
func ( o HeaderCallOption ) before ( c * callInfo ) error { return nil }
2020-06-25 22:42:41 +02:00
func ( o HeaderCallOption ) after ( c * callInfo , attempt * csAttempt ) {
* o . HeaderAddr , _ = attempt . s . Header ( )
2019-11-07 20:05:39 +01:00
}
// Trailer returns a CallOptions that retrieves the trailer metadata
// for a unary RPC.
func Trailer ( md * metadata . MD ) CallOption {
return TrailerCallOption { TrailerAddr : md }
}
// TrailerCallOption is a CallOption for collecting response trailer metadata.
// The metadata field will be populated *after* the RPC completes.
2020-10-16 14:04:44 +02:00
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type TrailerCallOption struct {
TrailerAddr * metadata . MD
}
func ( o TrailerCallOption ) before ( c * callInfo ) error { return nil }
2020-06-25 22:42:41 +02:00
func ( o TrailerCallOption ) after ( c * callInfo , attempt * csAttempt ) {
* o . TrailerAddr = attempt . s . Trailer ( )
2019-11-07 20:05:39 +01:00
}
// Peer returns a CallOption that retrieves peer information for a unary RPC.
// The peer field will be populated *after* the RPC completes.
func Peer ( p * peer . Peer ) CallOption {
return PeerCallOption { PeerAddr : p }
}
// PeerCallOption is a CallOption for collecting the identity of the remote
// peer. The peer field will be populated *after* the RPC completes.
2020-10-16 14:04:44 +02:00
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type PeerCallOption struct {
PeerAddr * peer . Peer
}
func ( o PeerCallOption ) before ( c * callInfo ) error { return nil }
2020-06-25 22:42:41 +02:00
func ( o PeerCallOption ) after ( c * callInfo , attempt * csAttempt ) {
if x , ok := peer . FromContext ( attempt . s . Context ( ) ) ; ok {
* o . PeerAddr = * x
2019-11-07 20:05:39 +01:00
}
}
// WaitForReady configures the action to take when an RPC is attempted on broken
2021-07-07 15:10:33 +02:00
// connections or unreachable servers. If waitForReady is false and the
// connection is in the TRANSIENT_FAILURE state, the RPC will fail
2019-11-07 20:05:39 +01:00
// immediately. Otherwise, the RPC client will block the call until a
// connection is available (or the call is canceled or times out) and will
// retry the call if it fails due to a transient error. gRPC will not retry if
// data was written to the wire unless the server indicates it did not process
// the data. Please refer to
// https://github.com/grpc/grpc/blob/master/doc/wait-for-ready.md.
//
// By default, RPCs don't "wait for ready".
func WaitForReady ( waitForReady bool ) CallOption {
return FailFastCallOption { FailFast : ! waitForReady }
}
// FailFast is the opposite of WaitForReady.
//
// Deprecated: use WaitForReady.
func FailFast ( failFast bool ) CallOption {
return FailFastCallOption { FailFast : failFast }
}
// FailFastCallOption is a CallOption for indicating whether an RPC should fail
// fast or not.
2020-10-16 14:04:44 +02:00
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type FailFastCallOption struct {
FailFast bool
}
func ( o FailFastCallOption ) before ( c * callInfo ) error {
c . failFast = o . FailFast
return nil
}
2020-06-25 22:42:41 +02:00
func ( o FailFastCallOption ) after ( c * callInfo , attempt * csAttempt ) { }
2019-11-07 20:05:39 +01:00
2023-03-25 02:08:06 +01:00
// OnFinish returns a CallOption that configures a callback to be called when
// the call completes. The error passed to the callback is the status of the
// RPC, and may be nil. The onFinish callback provided will only be called once
// by gRPC. This is mainly used to be used by streaming interceptors, to be
// notified when the RPC completes along with information about the status of
// the RPC.
//
// # Experimental
//
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release.
func OnFinish ( onFinish func ( err error ) ) CallOption {
return OnFinishCallOption {
OnFinish : onFinish ,
}
}
// OnFinishCallOption is CallOption that indicates a callback to be called when
// the call completes.
//
// # Experimental
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
type OnFinishCallOption struct {
OnFinish func ( error )
}
func ( o OnFinishCallOption ) before ( c * callInfo ) error {
c . onFinish = append ( c . onFinish , o . OnFinish )
return nil
}
func ( o OnFinishCallOption ) after ( c * callInfo , attempt * csAttempt ) { }
2020-03-12 17:11:27 +01:00
// MaxCallRecvMsgSize returns a CallOption which sets the maximum message size
2023-02-08 17:55:14 +01:00
// in bytes the client can receive. If this is not set, gRPC uses the default
// 4MB.
2020-03-12 17:11:27 +01:00
func MaxCallRecvMsgSize ( bytes int ) CallOption {
return MaxRecvMsgSizeCallOption { MaxRecvMsgSize : bytes }
2019-11-07 20:05:39 +01:00
}
// MaxRecvMsgSizeCallOption is a CallOption that indicates the maximum message
2020-03-12 17:11:27 +01:00
// size in bytes the client can receive.
2020-10-16 14:04:44 +02:00
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type MaxRecvMsgSizeCallOption struct {
MaxRecvMsgSize int
}
func ( o MaxRecvMsgSizeCallOption ) before ( c * callInfo ) error {
c . maxReceiveMessageSize = & o . MaxRecvMsgSize
return nil
}
2020-06-25 22:42:41 +02:00
func ( o MaxRecvMsgSizeCallOption ) after ( c * callInfo , attempt * csAttempt ) { }
2019-11-07 20:05:39 +01:00
2020-03-12 17:11:27 +01:00
// MaxCallSendMsgSize returns a CallOption which sets the maximum message size
2023-02-08 17:55:14 +01:00
// in bytes the client can send. If this is not set, gRPC uses the default
// `math.MaxInt32`.
2020-03-12 17:11:27 +01:00
func MaxCallSendMsgSize ( bytes int ) CallOption {
return MaxSendMsgSizeCallOption { MaxSendMsgSize : bytes }
2019-11-07 20:05:39 +01:00
}
// MaxSendMsgSizeCallOption is a CallOption that indicates the maximum message
2020-03-12 17:11:27 +01:00
// size in bytes the client can send.
2020-10-16 14:04:44 +02:00
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type MaxSendMsgSizeCallOption struct {
MaxSendMsgSize int
}
func ( o MaxSendMsgSizeCallOption ) before ( c * callInfo ) error {
c . maxSendMessageSize = & o . MaxSendMsgSize
return nil
}
2020-06-25 22:42:41 +02:00
func ( o MaxSendMsgSizeCallOption ) after ( c * callInfo , attempt * csAttempt ) { }
2019-11-07 20:05:39 +01:00
// PerRPCCredentials returns a CallOption that sets credentials.PerRPCCredentials
// for a call.
func PerRPCCredentials ( creds credentials . PerRPCCredentials ) CallOption {
return PerRPCCredsCallOption { Creds : creds }
}
// PerRPCCredsCallOption is a CallOption that indicates the per-RPC
// credentials to use for the call.
2020-10-16 14:04:44 +02:00
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type PerRPCCredsCallOption struct {
Creds credentials . PerRPCCredentials
}
func ( o PerRPCCredsCallOption ) before ( c * callInfo ) error {
c . creds = o . Creds
return nil
}
2020-06-25 22:42:41 +02:00
func ( o PerRPCCredsCallOption ) after ( c * callInfo , attempt * csAttempt ) { }
2019-11-07 20:05:39 +01:00
// UseCompressor returns a CallOption which sets the compressor used when
// sending the request. If WithCompressor is also set, UseCompressor has
// higher priority.
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
func UseCompressor ( name string ) CallOption {
return CompressorCallOption { CompressorType : name }
}
// CompressorCallOption is a CallOption that indicates the compressor to use.
2020-10-16 14:04:44 +02:00
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type CompressorCallOption struct {
CompressorType string
}
func ( o CompressorCallOption ) before ( c * callInfo ) error {
c . compressorType = o . CompressorType
return nil
}
2020-06-25 22:42:41 +02:00
func ( o CompressorCallOption ) after ( c * callInfo , attempt * csAttempt ) { }
2019-11-07 20:05:39 +01:00
// CallContentSubtype returns a CallOption that will set the content-subtype
// for a call. For example, if content-subtype is "json", the Content-Type over
// the wire will be "application/grpc+json". The content-subtype is converted
// to lowercase before being included in Content-Type. See Content-Type on
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
// more details.
//
// If ForceCodec is not also used, the content-subtype will be used to look up
// the Codec to use in the registry controlled by RegisterCodec. See the
// documentation on RegisterCodec for details on registration. The lookup of
// content-subtype is case-insensitive. If no such Codec is found, the call
// will result in an error with code codes.Internal.
//
// If ForceCodec is also used, that Codec will be used for all request and
// response messages, with the content-subtype set to the given contentSubtype
// here for requests.
func CallContentSubtype ( contentSubtype string ) CallOption {
return ContentSubtypeCallOption { ContentSubtype : strings . ToLower ( contentSubtype ) }
}
// ContentSubtypeCallOption is a CallOption that indicates the content-subtype
// used for marshaling messages.
2020-10-16 14:04:44 +02:00
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type ContentSubtypeCallOption struct {
ContentSubtype string
}
func ( o ContentSubtypeCallOption ) before ( c * callInfo ) error {
c . contentSubtype = o . ContentSubtype
return nil
}
2020-06-25 22:42:41 +02:00
func ( o ContentSubtypeCallOption ) after ( c * callInfo , attempt * csAttempt ) { }
2019-11-07 20:05:39 +01:00
2021-05-20 17:38:21 +02:00
// ForceCodec returns a CallOption that will set codec to be used for all
// request and response messages for a call. The result of calling Name() will
// be used as the content-subtype after converting to lowercase, unless
// CallContentSubtype is also used.
2019-11-07 20:05:39 +01:00
//
// See Content-Type on
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
// more details. Also see the documentation on RegisterCodec and
// CallContentSubtype for more details on the interaction between Codec and
// content-subtype.
//
// This function is provided for advanced users; prefer to use only
// CallContentSubtype to select a registered codec instead.
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
func ForceCodec ( codec encoding . Codec ) CallOption {
return ForceCodecCallOption { Codec : codec }
}
// ForceCodecCallOption is a CallOption that indicates the codec used for
// marshaling messages.
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type ForceCodecCallOption struct {
Codec encoding . Codec
}
func ( o ForceCodecCallOption ) before ( c * callInfo ) error {
c . codec = o . Codec
return nil
}
2020-06-25 22:42:41 +02:00
func ( o ForceCodecCallOption ) after ( c * callInfo , attempt * csAttempt ) { }
2019-11-07 20:05:39 +01:00
// CallCustomCodec behaves like ForceCodec, but accepts a grpc.Codec instead of
// an encoding.Codec.
//
// Deprecated: use ForceCodec instead.
func CallCustomCodec ( codec Codec ) CallOption {
return CustomCodecCallOption { Codec : codec }
}
// CustomCodecCallOption is a CallOption that indicates the codec used for
// marshaling messages.
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type CustomCodecCallOption struct {
Codec Codec
}
func ( o CustomCodecCallOption ) before ( c * callInfo ) error {
c . codec = o . Codec
return nil
}
2020-06-25 22:42:41 +02:00
func ( o CustomCodecCallOption ) after ( c * callInfo , attempt * csAttempt ) { }
2019-11-07 20:05:39 +01:00
// MaxRetryRPCBufferSize returns a CallOption that limits the amount of memory
// used for buffering this RPC's requests for retry purposes.
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This API is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
func MaxRetryRPCBufferSize ( bytes int ) CallOption {
return MaxRetryRPCBufferSizeCallOption { bytes }
}
// MaxRetryRPCBufferSizeCallOption is a CallOption indicating the amount of
// memory to be used for caching this RPC for retry purposes.
2020-10-16 14:04:44 +02:00
//
2022-11-22 10:26:16 +01:00
// # Experimental
2020-10-16 14:04:44 +02:00
//
// Notice: This type is EXPERIMENTAL and may be changed or removed in a
// later release.
2019-11-07 20:05:39 +01:00
type MaxRetryRPCBufferSizeCallOption struct {
MaxRetryRPCBufferSize int
}
func ( o MaxRetryRPCBufferSizeCallOption ) before ( c * callInfo ) error {
c . maxRetryRPCBufferSize = o . MaxRetryRPCBufferSize
return nil
}
2020-06-25 22:42:41 +02:00
func ( o MaxRetryRPCBufferSizeCallOption ) after ( c * callInfo , attempt * csAttempt ) { }
2019-11-07 20:05:39 +01:00
// The format of the payload: compressed or not?
type payloadFormat uint8
const (
compressionNone payloadFormat = 0 // no compression
compressionMade payloadFormat = 1 // compressed
)
// parser reads complete gRPC messages from the underlying reader.
type parser struct {
// r is the underlying reader.
// See the comment on recvMsg for the permissible
// error types.
r io . Reader
// The header of a gRPC message. Find more detail at
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md
header [ 5 ] byte
2023-07-27 00:05:15 +02:00
// recvBufferPool is the pool of shared receive buffers.
recvBufferPool SharedBufferPool
2019-11-07 20:05:39 +01:00
}
// recvMsg reads a complete gRPC message from the stream.
//
// It returns the message and its payload (compression/encoding)
// format. The caller owns the returned msg memory.
//
// If there is an error, possible values are:
2022-11-22 10:26:16 +01:00
// - io.EOF, when no messages remain
// - io.ErrUnexpectedEOF
// - of type transport.ConnectionError
// - an error from the status package
//
2019-11-07 20:05:39 +01:00
// No other error values or types must be returned, which also means
// that the underlying io.Reader must not return an incompatible
// error.
func ( p * parser ) recvMsg ( maxReceiveMessageSize int ) ( pf payloadFormat , msg [ ] byte , err error ) {
if _ , err := p . r . Read ( p . header [ : ] ) ; err != nil {
return 0 , nil , err
}
pf = payloadFormat ( p . header [ 0 ] )
length := binary . BigEndian . Uint32 ( p . header [ 1 : ] )
if length == 0 {
return pf , nil , nil
}
if int64 ( length ) > int64 ( maxInt ) {
return 0 , nil , status . Errorf ( codes . ResourceExhausted , "grpc: received message larger than max length allowed on current machine (%d vs. %d)" , length , maxInt )
}
if int ( length ) > maxReceiveMessageSize {
return 0 , nil , status . Errorf ( codes . ResourceExhausted , "grpc: received message larger than max (%d vs. %d)" , length , maxReceiveMessageSize )
}
2023-07-27 00:05:15 +02:00
msg = p . recvBufferPool . Get ( int ( length ) )
2019-11-07 20:05:39 +01:00
if _ , err := p . r . Read ( msg ) ; err != nil {
if err == io . EOF {
err = io . ErrUnexpectedEOF
}
return 0 , nil , err
}
return pf , msg , nil
}
// encode serializes msg and returns a buffer containing the message, or an
// error if it is too large to be transmitted by grpc. If msg is nil, it
// generates an empty message.
2023-09-07 12:34:14 +02:00
func encode ( c baseCodec , msg any ) ( [ ] byte , error ) {
2019-11-07 20:05:39 +01:00
if msg == nil { // NOTE: typed nils will not be caught by this check
return nil , nil
}
b , err := c . Marshal ( msg )
if err != nil {
return nil , status . Errorf ( codes . Internal , "grpc: error while marshaling: %v" , err . Error ( ) )
}
if uint ( len ( b ) ) > math . MaxUint32 {
return nil , status . Errorf ( codes . ResourceExhausted , "grpc: message too large (%d bytes)" , len ( b ) )
}
return b , nil
}
// compress returns the input bytes compressed by compressor or cp. If both
// compressors are nil, returns nil.
//
// TODO(dfawley): eliminate cp parameter by wrapping Compressor in an encoding.Compressor.
func compress ( in [ ] byte , cp Compressor , compressor encoding . Compressor ) ( [ ] byte , error ) {
if compressor == nil && cp == nil {
return nil , nil
}
wrapErr := func ( err error ) error {
return status . Errorf ( codes . Internal , "grpc: error while compressing: %v" , err . Error ( ) )
}
cbuf := & bytes . Buffer { }
if compressor != nil {
z , err := compressor . Compress ( cbuf )
if err != nil {
return nil , wrapErr ( err )
}
if _ , err := z . Write ( in ) ; err != nil {
return nil , wrapErr ( err )
}
if err := z . Close ( ) ; err != nil {
return nil , wrapErr ( err )
}
} else {
if err := cp . Do ( cbuf , in ) ; err != nil {
return nil , wrapErr ( err )
}
}
return cbuf . Bytes ( ) , nil
}
const (
payloadLen = 1
sizeLen = 4
headerLen = payloadLen + sizeLen
)
// msgHeader returns a 5-byte header for the message being transmitted and the
// payload, which is compData if non-nil or data otherwise.
func msgHeader ( data , compData [ ] byte ) ( hdr [ ] byte , payload [ ] byte ) {
hdr = make ( [ ] byte , headerLen )
if compData != nil {
hdr [ 0 ] = byte ( compressionMade )
data = compData
} else {
hdr [ 0 ] = byte ( compressionNone )
}
// Write length of payload into buf
binary . BigEndian . PutUint32 ( hdr [ payloadLen : ] , uint32 ( len ( data ) ) )
return hdr , data
}
2023-09-07 12:34:14 +02:00
func outPayload ( client bool , msg any , data , payload [ ] byte , t time . Time ) * stats . OutPayload {
2019-11-07 20:05:39 +01:00
return & stats . OutPayload {
2023-03-25 02:08:06 +01:00
Client : client ,
Payload : msg ,
Data : data ,
Length : len ( data ) ,
WireLength : len ( payload ) + headerLen ,
CompressedLength : len ( payload ) ,
SentTime : t ,
2019-11-07 20:05:39 +01:00
}
}
func checkRecvPayload ( pf payloadFormat , recvCompress string , haveCompressor bool ) * status . Status {
switch pf {
case compressionNone :
case compressionMade :
if recvCompress == "" || recvCompress == encoding . Identity {
return status . New ( codes . Internal , "grpc: compressed flag set with identity or empty encoding" )
}
if ! haveCompressor {
return status . Newf ( codes . Unimplemented , "grpc: Decompressor is not installed for grpc-encoding %q" , recvCompress )
}
default :
return status . Newf ( codes . Internal , "grpc: received unexpected payload format %d" , pf )
}
return nil
}
type payloadInfo struct {
2023-03-25 02:08:06 +01:00
compressedLength int // The compressed length got from wire.
2019-11-07 20:05:39 +01:00
uncompressedBytes [ ] byte
}
func recvAndDecompress ( p * parser , s * transport . Stream , dc Decompressor , maxReceiveMessageSize int , payInfo * payloadInfo , compressor encoding . Compressor ) ( [ ] byte , error ) {
2023-07-27 00:05:15 +02:00
pf , buf , err := p . recvMsg ( maxReceiveMessageSize )
2019-11-07 20:05:39 +01:00
if err != nil {
return nil , err
}
if payInfo != nil {
2023-07-27 00:05:15 +02:00
payInfo . compressedLength = len ( buf )
2019-11-07 20:05:39 +01:00
}
if st := checkRecvPayload ( pf , s . RecvCompress ( ) , compressor != nil || dc != nil ) ; st != nil {
return nil , st . Err ( )
}
var size int
if pf == compressionMade {
// To match legacy behavior, if the decompressor is set by WithDecompressor or RPCDecompressor,
// use this decompressor as the default.
if dc != nil {
2023-07-27 00:05:15 +02:00
buf , err = dc . Do ( bytes . NewReader ( buf ) )
size = len ( buf )
2019-11-07 20:05:39 +01:00
} else {
2023-07-27 00:05:15 +02:00
buf , size , err = decompress ( compressor , buf , maxReceiveMessageSize )
2019-11-07 20:05:39 +01:00
}
if err != nil {
2023-02-08 17:55:14 +01:00
return nil , status . Errorf ( codes . Internal , "grpc: failed to decompress the received message: %v" , err )
2019-11-07 20:05:39 +01:00
}
2021-12-20 11:06:52 +01:00
if size > maxReceiveMessageSize {
// TODO: Revisit the error code. Currently keep it consistent with java
// implementation.
return nil , status . Errorf ( codes . ResourceExhausted , "grpc: received message after decompression larger than max (%d vs. %d)" , size , maxReceiveMessageSize )
}
2019-11-07 20:05:39 +01:00
}
2023-07-27 00:05:15 +02:00
return buf , nil
2019-11-07 20:05:39 +01:00
}
// Using compressor, decompress d, returning data and size.
// Optionally, if data will be over maxReceiveMessageSize, just return the size.
func decompress ( compressor encoding . Compressor , d [ ] byte , maxReceiveMessageSize int ) ( [ ] byte , int , error ) {
dcReader , err := compressor . Decompress ( bytes . NewReader ( d ) )
if err != nil {
return nil , 0 , err
}
if sizer , ok := compressor . ( interface {
DecompressedSize ( compressedBytes [ ] byte ) int
} ) ; ok {
if size := sizer . DecompressedSize ( d ) ; size >= 0 {
if size > maxReceiveMessageSize {
return nil , size , nil
}
// size is used as an estimate to size the buffer, but we
// will read more data if available.
// +MinRead so ReadFrom will not reallocate if size is correct.
buf := bytes . NewBuffer ( make ( [ ] byte , 0 , size + bytes . MinRead ) )
bytesRead , err := buf . ReadFrom ( io . LimitReader ( dcReader , int64 ( maxReceiveMessageSize ) + 1 ) )
return buf . Bytes ( ) , int ( bytesRead ) , err
}
}
// Read from LimitReader with limit max+1. So if the underlying
// reader is over limit, the result will be bigger than max.
2023-02-08 17:55:14 +01:00
d , err = io . ReadAll ( io . LimitReader ( dcReader , int64 ( maxReceiveMessageSize ) + 1 ) )
2019-11-07 20:05:39 +01:00
return d , len ( d ) , err
}
// For the two compressor parameters, both should not be set, but if they are,
// dc takes precedence over compressor.
// TODO(dfawley): wrap the old compressor/decompressor using the new API?
2023-09-07 12:34:14 +02:00
func recv ( p * parser , c baseCodec , s * transport . Stream , dc Decompressor , m any , maxReceiveMessageSize int , payInfo * payloadInfo , compressor encoding . Compressor ) error {
2023-07-27 00:05:15 +02:00
buf , err := recvAndDecompress ( p , s , dc , maxReceiveMessageSize , payInfo , compressor )
2019-11-07 20:05:39 +01:00
if err != nil {
return err
}
2023-07-27 00:05:15 +02:00
if err := c . Unmarshal ( buf , m ) ; err != nil {
2023-02-08 17:55:14 +01:00
return status . Errorf ( codes . Internal , "grpc: failed to unmarshal the received message: %v" , err )
2019-11-07 20:05:39 +01:00
}
if payInfo != nil {
2023-07-27 00:05:15 +02:00
payInfo . uncompressedBytes = buf
} else {
p . recvBufferPool . Put ( & buf )
2019-11-07 20:05:39 +01:00
}
return nil
}
// Information about RPC
type rpcInfo struct {
failfast bool
preloaderInfo * compressorInfo
}
// Information about Preloader
// Responsible for storing codec, and compressors
// If stream (s) has context s.Context which stores rpcInfo that has non nil
// pointers to codec, and compressors, then we can use preparedMsg for Async message prep
// and reuse marshalled bytes
type compressorInfo struct {
codec baseCodec
cp Compressor
comp encoding . Compressor
}
type rpcInfoContextKey struct { }
func newContextWithRPCInfo ( ctx context . Context , failfast bool , codec baseCodec , cp Compressor , comp encoding . Compressor ) context . Context {
return context . WithValue ( ctx , rpcInfoContextKey { } , & rpcInfo {
failfast : failfast ,
preloaderInfo : & compressorInfo {
codec : codec ,
cp : cp ,
comp : comp ,
} ,
} )
}
func rpcInfoFromContext ( ctx context . Context ) ( s * rpcInfo , ok bool ) {
s , ok = ctx . Value ( rpcInfoContextKey { } ) . ( * rpcInfo )
return
}
// Code returns the error code for err if it was produced by the rpc system.
// Otherwise, it returns codes.Unknown.
//
// Deprecated: use status.Code instead.
func Code ( err error ) codes . Code {
return status . Code ( err )
}
// ErrorDesc returns the error description of err if it was produced by the rpc system.
// Otherwise, it returns err.Error() or empty string when err is nil.
//
// Deprecated: use status.Convert and Message method instead.
func ErrorDesc ( err error ) string {
return status . Convert ( err ) . Message ( )
}
// Errorf returns an error containing an error code and a description;
// Errorf returns nil if c is OK.
//
// Deprecated: use status.Errorf instead.
2023-09-07 12:34:14 +02:00
func Errorf ( c codes . Code , format string , a ... any ) error {
2019-11-07 20:05:39 +01:00
return status . Errorf ( c , format , a ... )
}
2023-09-07 12:34:14 +02:00
var errContextCanceled = status . Error ( codes . Canceled , context . Canceled . Error ( ) )
var errContextDeadline = status . Error ( codes . DeadlineExceeded , context . DeadlineExceeded . Error ( ) )
2019-11-07 20:05:39 +01:00
// toRPCErr converts an error into an error from the status package.
func toRPCErr ( err error ) error {
2021-08-13 12:03:03 +02:00
switch err {
case nil , io . EOF :
2019-11-07 20:05:39 +01:00
return err
2021-08-13 12:03:03 +02:00
case context . DeadlineExceeded :
2023-09-07 12:34:14 +02:00
return errContextDeadline
2021-08-13 12:03:03 +02:00
case context . Canceled :
2023-09-07 12:34:14 +02:00
return errContextCanceled
2021-08-13 12:03:03 +02:00
case io . ErrUnexpectedEOF :
2019-11-07 20:05:39 +01:00
return status . Error ( codes . Internal , err . Error ( ) )
}
2021-08-13 12:03:03 +02:00
2019-11-07 20:05:39 +01:00
switch e := err . ( type ) {
case transport . ConnectionError :
return status . Error ( codes . Unavailable , e . Desc )
2021-08-13 12:03:03 +02:00
case * transport . NewStreamError :
return toRPCErr ( e . Err )
}
if _ , ok := status . FromError ( err ) ; ok {
return err
2019-11-07 20:05:39 +01:00
}
2021-08-13 12:03:03 +02:00
2019-11-07 20:05:39 +01:00
return status . Error ( codes . Unknown , err . Error ( ) )
}
// setCallInfoCodec should only be called after CallOptions have been applied.
func setCallInfoCodec ( c * callInfo ) error {
if c . codec != nil {
2021-05-20 17:38:21 +02:00
// codec was already set by a CallOption; use it, but set the content
// subtype if it is not set.
if c . contentSubtype == "" {
// c.codec is a baseCodec to hide the difference between grpc.Codec and
// encoding.Codec (Name vs. String method name). We only support
// setting content subtype from encoding.Codec to avoid a behavior
// change with the deprecated version.
if ec , ok := c . codec . ( encoding . Codec ) ; ok {
c . contentSubtype = strings . ToLower ( ec . Name ( ) )
}
}
2019-11-07 20:05:39 +01:00
return nil
}
if c . contentSubtype == "" {
// No codec specified in CallOptions; use proto by default.
c . codec = encoding . GetCodec ( proto . Name )
return nil
}
// c.contentSubtype is already lowercased in CallContentSubtype
c . codec = encoding . GetCodec ( c . contentSubtype )
if c . codec == nil {
return status . Errorf ( codes . Internal , "no codec registered for content-subtype %s" , c . contentSubtype )
}
return nil
}
// channelzData is used to store channelz related data for ClientConn, addrConn and Server.
// These fields cannot be embedded in the original structs (e.g. ClientConn), since to do atomic
// operation on int64 variable on 32-bit machine, user is responsible to enforce memory alignment.
// Here, by grouping those int64 fields inside a struct, we are enforcing the alignment.
type channelzData struct {
callsStarted int64
callsFailed int64
callsSucceeded int64
// lastCallStartedTime stores the timestamp that last call starts. It is of int64 type instead of
// time.Time since it's more costly to atomically update time.Time variable than int64 variable.
lastCallStartedTime int64
}
// The SupportPackageIsVersion variables are referenced from generated protocol
// buffer files to ensure compatibility with the gRPC version used. The latest
2020-09-10 08:49:55 +02:00
// support package version is 7.
2019-11-07 20:05:39 +01:00
//
2021-02-28 13:51:43 +01:00
// Older versions are kept for compatibility.
2019-11-07 20:05:39 +01:00
//
// These constants should not be referenced from any other code.
const (
SupportPackageIsVersion3 = true
SupportPackageIsVersion4 = true
SupportPackageIsVersion5 = true
2020-02-10 22:35:08 +01:00
SupportPackageIsVersion6 = true
2020-09-10 08:49:55 +02:00
SupportPackageIsVersion7 = true
2019-11-07 20:05:39 +01:00
)
const grpcUA = "grpc-go/" + Version