grpc

package
v0.6.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 31, 2023 License: Apache-2.0 Imports: 35 Imported by: 0

Documentation

Overview

The files in grpc package are forked from gRPC[github.com/grpc/grpc-go], and we keep the original Copyright[Copyright 2017 gRPC authors] and License of gRPC for those files. We also need to modify as we need, the modifications are Copyright of 2021 CloudWeGo Authors. Thanks for gRPC authors! Below is the source code information: 

Repo: github.com/grpc/grpc-go
Forked Version: v1.26.0

Package grpc defines and implements message oriented communication channel to complete various transactions (e.g., an RPC). It is meant for grpc-internal usage and is not intended to be imported directly by users.

Index

Constants

View Source 
const (

	// Infinity means unset duration
	Infinity = time.Duration(math.MaxInt64)
)
View Source 
const (
	// KeepaliveMinPingTime is the minimum ping interval.
	// This must be 10s by default, but tests may wish to set it lower for convenience.
	KeepaliveMinPingTime = 10 * time.Second
)

Variables

View Source 
var (
	// ErrIllegalHeaderWrite indicates that setting header is illegal because of
	// the stream's state.
	ErrIllegalHeaderWrite = errors.New("transport: the stream is done or WriteHeader was already called")
	// ErrHeaderListSizeLimitViolation indicates that the header list size is larger
	// than the limit set by peer.
	ErrHeaderListSizeLimitViolation = errors.New("transport: trying to send header list size larger than the limit set by peer")
)
View Source 
var (
	// ClientPreface http2 preface message
	ClientPreface = []byte(http2.ClientPreface)
	// ClientPrefaceLen preface length
	ClientPrefaceLen = len(ClientPreface)

	// HTTPStatusConvTab is the HTTP status code to gRPC error code conversion table.
	HTTPStatusConvTab = map[int]codes.Code{

		http.StatusBadRequest: codes.Internal,

		http.StatusUnauthorized: codes.Unauthenticated,

		http.StatusForbidden: codes.PermissionDenied,

		http.StatusNotFound: codes.Unimplemented,

		http.StatusTooManyRequests: codes.Unavailable,

		http.StatusBadGateway: codes.Unavailable,

		http.StatusServiceUnavailable: codes.Unavailable,

		http.StatusGatewayTimeout: codes.Unavailable,
	}
)
View Source 
var (
	// ErrConnClosing indicates that the transport is closing.
	ErrConnClosing = connectionErrorf(true, nil, "transport is closing")
)

Functions

func ContextErr 

func ContextErr(err error) error

ContextErr converts the error from context package into a status error.

func StreamWrite 

func StreamWrite(s *Stream, buffer *bytes.Buffer)

StreamWrite only used for unit test

Types

type CallHdr 

type CallHdr struct {
	// Host specifies the peer's host.
	Host string

	// Method specifies the operation to perform.
	Method string

	// SendCompress specifies the compression algorithm applied on
	// outbound message.
	SendCompress string

	// ContentSubtype specifies the content-subtype for a request. For example, a
	// content-subtype of "proto" will result in a content-type of
	// "application/grpc+proto". The value of ContentSubtype must be all
	// lowercase, otherwise the behavior is undefined. See
	// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
	// for more details.
	ContentSubtype string

	PreviousAttempts int // value of grpc-previous-rpc-attempts header to set
}

CallHdr carries the information of a particular RPC.

type ClientKeepalive 

type ClientKeepalive struct {
	// After a duration of this time if the client doesn't see any activity it
	// pings the server to see if the transport is still alive.
	// If set below 10s, a minimum value of 10s will be used instead.
	Time time.Duration // The current default value is infinity.
	// After having pinged for keepalive check, the client waits for a duration
	// of Timeout and if no activity is seen even after that the connection is
	// closed.
	Timeout time.Duration // The current default value is 20 seconds.
	// If true, client sends keepalive pings even with no active RPCs. If false,
	// when there are no active RPCs, Time and Timeout will be ignored and no
	// keepalive pings will be sent.
	PermitWithoutStream bool // false by default.
}

ClientKeepalive is used to set keepalive parameters on the client-side. These configure how the client will actively probe to notice when a connection is broken and send pings so intermediaries will be aware of the liveness of the connection. Make sure these parameters are set in coordination with the keepalive policy on the server, as incompatible settings can result in closing of connection.

type ClientTransport 

type ClientTransport interface {
	// Close tears down this transport. Once it returns, the transport
	// should not be accessed any more. The caller must make sure this
	// is called only once.
	Close() error

	// GracefulClose starts to tear down the transport: the transport will stop
	// accepting new RPCs and NewStream will return error. Once all streams are
	// finished, the transport will close.
	//
	// It does not block.
	GracefulClose()

	// Write sends the data for the given stream. A nil stream indicates
	// the write is to be performed on the transport as a whole.
	Write(s *Stream, hdr, data []byte, opts *Options) error

	// NewStream creates a Stream for an RPC.
	NewStream(ctx context.Context, callHdr *CallHdr) (*Stream, error)

	// CloseStream clears the footprint of a stream when the stream is
	// not needed any more. The err indicates the error incurred when
	// CloseStream is called. Must be called when a stream is finished
	// unless the associated transport is closing.
	CloseStream(stream *Stream, err error)

	// Error returns a channel that is closed when some I/O error
	// happens. Typically the caller should have a goroutine to monitor
	// this in order to take action (e.g., close the current transport
	// and create a new one) in error case. It should not return nil
	// once the transport is initiated.
	Error() <-chan struct{}

	// GoAway returns a channel that is closed when ClientTransport
	// receives the draining signal from the server (e.g., GOAWAY frame in
	// HTTP/2).
	GoAway() <-chan struct{}

	// GetGoAwayReason returns the reason why GoAway frame was received.
	GetGoAwayReason() GoAwayReason

	// RemoteAddr returns the remote network address.
	RemoteAddr() net.Addr
	LocalAddr() net.Addr
}

ClientTransport is the common interface for all gRPC client-side transport implementations.

func NewClientTransport 

func NewClientTransport(ctx context.Context, conn net.Conn, opts ConnectOptions,
	remoteService string, onGoAway func(GoAwayReason), onClose func(),
) (ClientTransport, error)

NewClientTransport establishes the transport with the required ConnectOptions and returns it to the caller.

type ConnectOptions 

type ConnectOptions struct {
	// KeepaliveParams stores the keepalive parameters.
	KeepaliveParams ClientKeepalive
	// InitialWindowSize sets the initial window size for a stream.
	InitialWindowSize uint32
	// InitialConnWindowSize sets the initial window size for a connection.
	InitialConnWindowSize uint32
	// WriteBufferSize sets the size of write buffer which in turn determines how much data can be batched before it's written on the wire.
	WriteBufferSize uint32
	// ReadBufferSize sets the size of read buffer, which in turn determines how much data can be read at most for one read syscall.
	ReadBufferSize uint32
	// MaxHeaderListSize sets the max (uncompressed) size of header list that is prepared to be received.
	MaxHeaderListSize *uint32
	// ShortConn indicates whether the connection will be reused from grpc conn pool
	ShortConn bool
	// TLSConfig
	TLSConfig *tls.Config
}

ConnectOptions covers all relevant options for communicating with the server.

type ConnectionError 

type ConnectionError struct {
	Desc string
	// contains filtered or unexported fields
}

ConnectionError is an error that results in the termination of the entire connection and the retry of all the active streams.

func (ConnectionError) Error 

func (e ConnectionError) Error() string

func (ConnectionError) Origin 

func (e ConnectionError) Origin() error

Origin returns the original error of this connection error.

func (ConnectionError) Temporary 

func (e ConnectionError) Temporary() bool

Temporary indicates if this connection error is temporary or fatal.

type EnforcementPolicy 

type EnforcementPolicy struct {
	// MinTime is the minimum amount of time a client should wait before sending
	// a keepalive ping.
	MinTime time.Duration // The current default value is 5 minutes.
	// If true, server allows keepalive pings even when there are no active
	// streams(RPCs). If false, and client sends ping when there are no active
	// streams, server will send GOAWAY and close the connection.
	PermitWithoutStream bool // false by default.
}

EnforcementPolicy is used to set keepalive enforcement policy on the server-side. Server will close connection with a client that violates this policy.

type GoAwayReason 

type GoAwayReason uint8

GoAwayReason contains the reason for the GoAway frame received. 

const (
	// GoAwayInvalid indicates that no GoAway frame is received.
	GoAwayInvalid GoAwayReason = 0
	// GoAwayNoReason is the default value when GoAway frame is received.
	GoAwayNoReason GoAwayReason = 1
	// GoAwayTooManyPings indicates that a GoAway frame with
	// ErrCodeEnhanceYourCalm was received and that the debug data said
	// "too_many_pings".
	GoAwayTooManyPings GoAwayReason = 2
)

type IsActive 

type IsActive interface {
	IsActive() bool
}

IsActive is the interface that exposing the underlying connection's active status.

type Options 

type Options struct {
	// Last indicates whether this write is the last piece for
	// this stream.
	Last bool
}

Options provides additional hints and information for message transmission.

type ServerConfig 

type ServerConfig struct {
	MaxStreams                 uint32
	KeepaliveParams            ServerKeepalive
	KeepaliveEnforcementPolicy EnforcementPolicy
	InitialWindowSize          uint32
	InitialConnWindowSize      uint32
	WriteBufferSize            uint32
	ReadBufferSize             uint32
	MaxHeaderListSize          *uint32
}

ServerConfig consists of all the configurations to establish a server transport.

func DefaultServerConfig 

func DefaultServerConfig() *ServerConfig

type ServerKeepalive 

type ServerKeepalive struct {
	// MaxConnectionIdle is a duration for the amount of time after which an
	// idle connection would be closed by sending a GoAway. Idleness duration is
	// defined since the most recent time the number of outstanding RPCs became
	// zero or the connection establishment.
	MaxConnectionIdle time.Duration // The current default value is infinity.
	// MaxConnectionAge is a duration for the maximum amount of time a
	// connection may exist before it will be closed by sending a GoAway. A
	// random jitter of +/-10% will be added to MaxConnectionAge to spread out
	// connection storms.
	MaxConnectionAge time.Duration // The current default value is infinity.
	// MaxConnectionAgeGrace is an additive period after MaxConnectionAge after
	// which the connection will be forcibly closed.
	MaxConnectionAgeGrace time.Duration // The current default value is infinity.
	// After a duration of this time if the server doesn't see any activity it
	// pings the client to see if the transport is still alive.
	// If set below 1s, a minimum value of 1s will be used instead.
	Time time.Duration // The current default value is 2 hours.
	// After having pinged for keepalive check, the server waits for a duration
	// of Timeout and if no activity is seen even after that the connection is
	// closed.
	Timeout time.Duration // The current default value is 20 seconds.
}

ServerKeepalive is used to set keepalive and max-age parameters on the server-side.

type ServerTransport 

type ServerTransport interface {
	// HandleStreams receives incoming streams using the given handler.
	HandleStreams(func(*Stream), func(context.Context, string) context.Context)

	// WriteHeader sends the header metadata for the given stream.
	// WriteHeader may not be called on all streams.
	WriteHeader(s *Stream, md metadata.MD) error

	// Write sends the data for the given stream.
	// Write may not be called on all streams.
	Write(s *Stream, hdr, data []byte, opts *Options) error

	// WriteStatus sends the status of a stream to the client.  WriteStatus is
	// the final call made on a stream and always occurs.
	WriteStatus(s *Stream, st *status.Status) error

	// Close tears down the transport. Once it is called, the transport
	// should not be accessed any more. All the pending streams and their
	// handlers will be terminated asynchronously.
	Close() error

	// RemoteAddr returns the remote network address.
	RemoteAddr() net.Addr
	LocalAddr() net.Addr

	// Drain notifies the client this ServerTransport stops accepting new RPCs.
	Drain()
}

ServerTransport is the common interface for all gRPC server-side transport implementations.

Methods may be called concurrently from multiple goroutines, but Write methods for a given Stream will be called serially.

func NewServerTransport 

func NewServerTransport(ctx context.Context, conn net.Conn, cfg *ServerConfig) (ServerTransport, error)

NewServerTransport creates a ServerTransport with conn or non-nil error if it fails.

type Stream 

type Stream struct {
	// contains filtered or unexported fields
}

Stream represents an RPC in the transport layer.

func CreateStream 

func CreateStream(id uint32, requestRead func(i int)) *Stream

CreateStream only used for unit test. Create an independent stream out of http2client / http2server

func (*Stream) BizStatusErr 

func (s *Stream) BizStatusErr() kerrors.BizStatusErrorIface

func (*Stream) BytesReceived 

func (s *Stream) BytesReceived() bool

BytesReceived indicates whether any bytes have been received on this stream.

func (*Stream) ContentSubtype 

func (s *Stream) ContentSubtype() string

ContentSubtype returns the content-subtype for a request. For example, a content-subtype of "proto" will result in a content-type of "application/grpc+proto". This will always be lowercase. See https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for more details.

func (*Stream) Context 

func (s *Stream) Context() context.Context

Context returns the context of the stream.

func (*Stream) Done 

func (s *Stream) Done() <-chan struct{}

Done returns a channel which is closed when it receives the final status from the server.

func (*Stream) Header 

func (s *Stream) Header() (metadata.MD, error)

Header returns the header metadata of the stream.

On client side, it acquires the key-value pairs of header metadata once it is available. It blocks until i) the metadata is ready or ii) there is no header metadata or iii) the stream is canceled/expired.

On server side, it returns the out header after t.WriteHeader is called. It does not block and must not be called until after WriteHeader.

func (*Stream) Method 

func (s *Stream) Method() string

Method returns the method for the stream.

func (*Stream) Read 

func (s *Stream) Read(p []byte) (n int, err error)

Read reads all p bytes from the wire for this stream.

func (*Stream) RecvCompress 

func (s *Stream) RecvCompress() string

RecvCompress returns the compression algorithm applied to the inbound message. It is empty string if there is no compression applied.

func (*Stream) SendHeader 

func (s *Stream) SendHeader(md metadata.MD) error

SendHeader sends the given header metadata. The given metadata is combined with any metadata set by previous calls to SetHeader and then written to the transport stream.

func (*Stream) SetBizStatusErr 

func (s *Stream) SetBizStatusErr(bizStatusErr kerrors.BizStatusErrorIface)

func (*Stream) SetHeader 

func (s *Stream) SetHeader(md metadata.MD) error

SetHeader sets the header metadata. This can be called multiple times. Server side only. This should not be called in parallel to other data writes.

func (*Stream) SetSendCompress 

func (s *Stream) SetSendCompress(str string)

SetSendCompress sets the compression algorithm to the stream.

func (*Stream) SetTrailer 

func (s *Stream) SetTrailer(md metadata.MD) error

SetTrailer sets the trailer metadata which will be sent with the RPC status by the server. This can be called multiple times. Server side only. This should not be called parallel to other data writes.

func (*Stream) Status 

func (s *Stream) Status() *status.Status

Status returns the status received from the server. Status can be read safely only after the stream has ended, that is, after Done() is closed.

func (*Stream) Trailer 

func (s *Stream) Trailer() metadata.MD

Trailer returns the cached trailer metedata. Note that if it is not called after the entire stream is done, it could return an empty MD. Client side only. It can be safely read only after stream has ended that is either read or write have returned io.EOF.

func (*Stream) TrailersOnly 

func (s *Stream) TrailersOnly() bool

TrailersOnly blocks until a header or trailers-only frame is received and then returns true if the stream was trailers-only. If the stream ends before headers are received, returns true, nil. Client-side only.

func (*Stream) Unprocessed 

func (s *Stream) Unprocessed() bool

Unprocessed indicates whether the server did not process this stream -- i.e. it sent a refused stream or GOAWAY including this stream ID.

Source Files

View all Source files

Directories

Path Synopsis
Package syscall provides functionalities that grpc uses to get low-level operating system stats/info.
 Package syscall provides functionalities that grpc uses to get low-level operating system stats/info.
leakcheck
Package leakcheck contains functions to check leaked goroutines.
 Package leakcheck contains functions to check leaked goroutines.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.