transport

package
v1.69.4 Latest Latest
Warning

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

 Go to latest Published: Jan 13, 2025 License: Apache-2.0 Imports: 46 Imported by: 0

Documentation

Overview

Package transport 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

This section is empty.

Variables

View Source 
var (
	// ErrIllegalHeaderWrite indicates that setting header is illegal because of
	// the stream's state.
	ErrIllegalHeaderWrite = status.Error(codes.Internal, "transport: SendHeader called multiple times")
	// ErrHeaderListSizeLimitViolation indicates that the header list size is larger
	// than the limit set by peer.
	ErrHeaderListSizeLimitViolation = status.Error(codes.Internal, "transport: trying to send header list size larger than the limit set by peer")
)
View Source 
var (
	// ErrConnClosing indicates that the transport is closing.
	ErrConnClosing = connectionErrorf(true, nil, "transport is closing")
)
View Source 
var (

	// 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 MaxStreamID = uint32(math.MaxInt32 * 3 / 4)

MaxStreamID is the upper bound for the stream ID before the current transport gracefully closes and new transport is created for subsequent RPCs. This is set to 75% of 2^31-1. Streams are identified with an unsigned 31-bit integer. It's exported so that tests can override it.

Functions

func ContextErr 

func ContextErr(err error) error

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

func GetConnection added in v1.41.0 

func GetConnection(ctx context.Context) net.Conn

GetConnection gets the connection from the context.

func SetConnection added in v1.60.0 

func SetConnection(ctx context.Context, conn net.Conn) context.Context

SetConnection adds the connection to the context to be able to get information about the destination ip and port for an incoming RPC. This also allows any unary or streaming interceptors to see the connection.

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

	// Creds specifies credentials.PerRPCCredentials for a call.
	Creds credentials.PerRPCCredentials

	// 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

	DoneFunc func() // called when the stream is finished
}

CallHdr carries the information of a particular RPC.

type ClientStream added in v1.69.0 

type ClientStream struct {
	*Stream // Embed for common stream functionality.
	// contains filtered or unexported fields
}

ClientStream implements streaming functionality for a gRPC client.

func (*ClientStream) BytesReceived added in v1.69.0 

func (s *ClientStream) BytesReceived() bool

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

func (*ClientStream) Close added in v1.69.0 

func (s *ClientStream) Close(err error)

Close closes the stream and popagates err to any readers.

func (*ClientStream) Done added in v1.69.0 

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

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

func (*ClientStream) Header added in v1.69.0 

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

Header returns the header metadata of the stream. 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.

func (*ClientStream) Read added in v1.69.0 

func (s *ClientStream) Read(n int) (mem.BufferSlice, error)

Read reads an n byte message from the input stream.

func (*ClientStream) RecvCompress added in v1.69.0 

func (s *ClientStream) RecvCompress() string

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

func (*ClientStream) Status added in v1.69.0 

func (s *ClientStream) 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 (*ClientStream) TrailersOnly added in v1.69.0 

func (s *ClientStream) 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.

func (*ClientStream) Unprocessed added in v1.69.0 

func (s *ClientStream) 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.

func (*ClientStream) Write added in v1.69.0 

func (s *ClientStream) Write(hdr []byte, data mem.BufferSlice, opts *WriteOptions) error

Write writes the hdr and data bytes to the output stream.

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(err 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()

	// NewStream creates a Stream for an RPC.
	NewStream(ctx context.Context, callHdr *CallHdr) (*ClientStream, 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, along
	// with a human readable string with debug info.
	GetGoAwayReason() (GoAwayReason, string)

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

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

func NewHTTP2Client added in v1.69.0 

func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts ConnectOptions, onClose func(GoAwayReason)) (_ ClientTransport, err error)

NewHTTP2Client constructs a connected ClientTransport to addr based on HTTP2 and starts to receive messages on it. Non-nil error returns if construction fails.

type ConnectOptions 

type ConnectOptions struct {
	// UserAgent is the application user agent.
	UserAgent string
	// Dialer specifies how to dial a network address.
	Dialer func(context.Context, string) (net.Conn, error)
	// FailOnNonTempDialError specifies if gRPC fails on non-temporary dial errors.
	FailOnNonTempDialError bool
	// PerRPCCredentials stores the PerRPCCredentials required to issue RPCs.
	PerRPCCredentials []credentials.PerRPCCredentials
	// TransportCredentials stores the Authenticator required to setup a client
	// connection. Only one of TransportCredentials and CredsBundle is non-nil.
	TransportCredentials credentials.TransportCredentials
	// CredsBundle is the credentials bundle to be used. Only one of
	// TransportCredentials and CredsBundle is non-nil.
	CredsBundle credentials.Bundle
	// KeepaliveParams stores the keepalive parameters.
	KeepaliveParams keepalive.ClientParameters
	// StatsHandlers stores the handler for stats.
	StatsHandlers []stats.Handler
	// InitialWindowSize sets the initial window size for a stream.
	InitialWindowSize int32
	// InitialConnWindowSize sets the initial window size for a connection.
	InitialConnWindowSize int32
	// 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 int
	// ReadBufferSize sets the size of read buffer, which in turn determines how much data can be read at most for one read syscall.
	ReadBufferSize int
	// SharedWriteBuffer indicates whether connections should reuse write buffer
	SharedWriteBuffer bool
	// ChannelzParent sets the addrConn id which initiated the creation of this client transport.
	ChannelzParent *channelz.SubChannel
	// MaxHeaderListSize sets the max (uncompressed) size of header list that is prepared to be received.
	MaxHeaderListSize *uint32
	// UseProxy specifies if a proxy should be used.
	UseProxy bool
	// The mem.BufferPool to use when reading/writing to the wire.
	BufferPool mem.BufferPool
}

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.

func (ConnectionError) Unwrap added in v1.45.0 

func (e ConnectionError) Unwrap() error

Unwrap returns the original error of this connection error or nil when the origin is nil.

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 NewStreamError added in v1.40.0 

type NewStreamError struct {
	Err error

	AllowTransparentRetry bool
}

NewStreamError wraps an error and reports additional information. Typically NewStream errors result in transparent retry, as they mean nothing went onto the wire. However, there are two notable exceptions:

  1. If the stream headers violate the max header list size allowed by the server. It's possible this could succeed on another transport, even if it's unlikely, but do not transparently retry.
  2. If the credentials errored when requesting their headers. In this case, it's possible a retry can fix the problem, but indefinitely transparently retrying is not appropriate as it is likely the credentials, if they can eventually succeed, would need I/O to do so.

func (NewStreamError) Error added in v1.40.0 

func (e NewStreamError) Error() string

type ServerConfig 

type ServerConfig struct {
	MaxStreams            uint32
	ConnectionTimeout     time.Duration
	Credentials           credentials.TransportCredentials
	InTapHandle           tap.ServerInHandle
	StatsHandlers         []stats.Handler
	KeepaliveParams       keepalive.ServerParameters
	KeepalivePolicy       keepalive.EnforcementPolicy
	InitialWindowSize     int32
	InitialConnWindowSize int32
	WriteBufferSize       int
	ReadBufferSize        int
	SharedWriteBuffer     bool
	ChannelzParent        *channelz.Server
	MaxHeaderListSize     *uint32
	HeaderTableSize       *uint32
	BufferPool            mem.BufferPool
}

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

type ServerStream added in v1.69.0 

type ServerStream struct {
	*Stream // Embed for common stream functionality.
	// contains filtered or unexported fields
}

ServerStream implements streaming functionality for a gRPC server.

func (*ServerStream) ClientAdvertisedCompressors added in v1.69.0 

func (s *ServerStream) ClientAdvertisedCompressors() []string

ClientAdvertisedCompressors returns the compressor names advertised by the client via grpc-accept-encoding header.

func (*ServerStream) ContentSubtype added in v1.69.0 

func (s *ServerStream) 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 (*ServerStream) Header added in v1.69.0 

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

Header returns the header metadata of the stream. It returns the out header after t.WriteHeader is called. It does not block and must not be called until after WriteHeader.

func (*ServerStream) HeaderWireLength added in v1.69.0 

func (s *ServerStream) HeaderWireLength() int

HeaderWireLength returns the size of the headers of the stream as received from the wire.

func (*ServerStream) Read added in v1.69.0 

func (s *ServerStream) Read(n int) (mem.BufferSlice, error)

Read reads an n byte message from the input stream.

func (*ServerStream) RecvCompress added in v1.69.0 

func (s *ServerStream) RecvCompress() string

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

func (*ServerStream) SendCompress added in v1.69.0 

func (s *ServerStream) SendCompress() string

SendCompress returns the send compressor name.

func (*ServerStream) SendHeader added in v1.69.0 

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

SendHeader sends the header metadata for the given stream.

func (*ServerStream) SetContext added in v1.69.0 

func (s *ServerStream) SetContext(ctx context.Context)

SetContext sets the context of the stream. This will be deleted once the stats handler callouts all move to gRPC layer.

func (*ServerStream) SetHeader added in v1.69.0 

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

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

func (*ServerStream) SetSendCompress added in v1.69.0 

func (s *ServerStream) SetSendCompress(name string) error

SetSendCompress sets the compression algorithm to the stream.

func (*ServerStream) SetTrailer added in v1.69.0 

func (s *ServerStream) 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. This should not be called parallel to other data writes.

func (*ServerStream) Write added in v1.69.0 

func (s *ServerStream) Write(hdr []byte, data mem.BufferSlice, opts *WriteOptions) error

Write writes the hdr and data bytes to the output stream.

func (*ServerStream) WriteStatus added in v1.69.0 

func (s *ServerStream) WriteStatus(st *status.Status) error

WriteStatus sends the status of a stream to the client. WriteStatus is the final call made on a stream and always occurs.

type ServerTransport 

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

	// 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(err error)

	// Peer returns the peer of the server transport.
	Peer() *peer.Peer

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

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 NewServerHandlerTransport 

func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request, stats []stats.Handler, bufferPool mem.BufferPool) (ServerTransport, error)

NewServerHandlerTransport returns a ServerTransport handling gRPC from inside an http.Handler, or writes an HTTP error to w and returns an error. It requires that the http Server supports HTTP/2.

func NewServerTransport 

func NewServerTransport(conn net.Conn, config *ServerConfig) (_ ServerTransport, err error)

NewServerTransport creates a http2 transport with conn and configuration options from config.

It returns a non-nil transport and a nil error on success. On failure, it returns a nil transport and a non-nil error. For a special case where the underlying conn gets closed before the client preface could be read, it returns a nil transport and a nil error.

type Stream 

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

Stream represents an RPC in the transport layer.

func (*Stream) Context 

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

Context returns the context of the stream.

func (*Stream) GoString 

func (s *Stream) GoString() string

GoString is implemented by Stream so context.String() won't race when printing %#v.

func (*Stream) Method 

func (s *Stream) Method() string

Method returns the method for the stream.

func (*Stream) ReadMessageHeader added in v1.69.0 

func (s *Stream) ReadMessageHeader(header []byte) (err error)

ReadMessageHeader reads data into the provided header slice from the stream. It first checks if there was an error during a previous read operation and returns it if present. It then requests a read operation for the length of the header. It continues to read from the stream until the entire header slice is filled or an error occurs. If an `io.EOF` error is encountered with partially read data, it is converted to `io.ErrUnexpectedEOF` to indicate an unexpected end of the stream. The method returns any error encountered during the read process or nil if the header was successfully read.

func (*Stream) Trailer 

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

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

type WriteOptions added in v1.69.0 

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

WriteOptions provides additional hints and information for message transmission.

Source Files

View all Source files

Directories

Path Synopsis
Package grpchttp2 defines HTTP/2 types and a framer API and implementation.
 Package grpchttp2 defines HTTP/2 types and a framer API and implementation.
Package networktype declares the network type to be used in the default dialer.
 Package networktype declares the network type to be used in the default dialer.

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.