transport

Documentation

Overview

Package transport defines and implements message oriented communication channel to complete various transactions (e.g., an RPC).

Index

• Variables
• type CallHdr
• type ClientTransport
  • func NewClientTransport(target string, opts *ConnectOptions) (ClientTransport, error)
• type ConnectOptions
• type ConnectionError
  • func ConnectionErrorf(format string, a ...interface{}) ConnectionError
  • func (e ConnectionError) Error() string
• type Options
• type ServerTransport
  • func NewServerTransport(protocol string, conn net.Conn, maxStreams uint32, ...) (ServerTransport, error)
• type Stream
  • func StreamFromContext(ctx context.Context) (s *Stream, ok bool)
  • func (s *Stream) Context() context.Context
  • func (s *Stream) Header() (metadata.MD, error)
  • func (s *Stream) Method() string
  • func (s *Stream) Read(p []byte) (n int, err error)
  • func (s *Stream) RecvCompress() string
  • func (s *Stream) ServerTransport() ServerTransport
  • func (s *Stream) SetSendCompress(str string)
  • func (s *Stream) SetTrailer(md metadata.MD) error
  • func (s *Stream) StatusCode() codes.Code
  • func (s *Stream) StatusDesc() string
  • func (s *Stream) TraceContext(tr trace.Trace)
  • func (s *Stream) Trailer() metadata.MD
• type StreamError
  • func ContextErr(err error) StreamError
  • func StreamErrorf(c codes.Code, format string, a ...interface{}) StreamError
  • func (e StreamError) Error() string

Variables

var ErrConnClosing = ConnectionError{Desc: "transport is closing"}

Define some common ConnectionErrors.

var ErrIllegalHeaderWrite = errors.New("transport: the stream is done or WriteHeader was already called")

ErrIllegalHeaderWrite indicates that setting header is illegal because of the stream's state.

var ErrIllegalTrailerSet = errors.New("transport: trailer has been set")

ErrIllegalTrailerSet indicates that the trailer has already been set or it is too late to do so.

Types

type CallHdr

type CallHdr struct {
	// Host specifies peer host.
	Host string
	// Method specifies the operation to perform.
	Method string
	// RecvCompress specifies the compression algorithm applied on inbound messages.
	RecvCompress string
	// SendCompress specifies the compression algorithm applied on outbound message.
	SendCompress string
	// Flush indicates if new stream command should be sent to the peer without
	// waiting for the first data. This is a hint though. The transport may modify
	// the flush decision for performance purpose.
	Flush bool
}

CallHdr carries the information of a particular RPC.

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

	// 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, 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{}
}

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

func NewClientTransport

func NewClientTransport(target string, opts *ConnectOptions) (ClientTransport, error)

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

type ConnectOptions

type ConnectOptions struct {
	// UserAgent is the application user agent.
	UserAgent string
	// Dialer specifies how to dial a network address.
	Dialer func(string, time.Duration) (net.Conn, error)
	// AuthOptions stores the credentials required to setup a client connection and/or issue RPCs.
	AuthOptions []credentials.Credentials
	// Timeout specifies the timeout for dialing a client connection.
	Timeout time.Duration
}

ConnectOptions covers all relevant options for dialing a server.

type ConnectionError

type ConnectionError struct {
	Desc string
}

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

func ConnectionErrorf

func ConnectionErrorf(format string, a ...interface{}) ConnectionError

ConnectionErrorf creates an ConnectionError with the specified error description.

func (ConnectionError) Error

func (e ConnectionError) Error() string

type Options

type Options struct {
	// Indicate whether it is the last piece for this stream.
	Last bool
	// The hint to transport impl whether the data could be buffered for
	// batching write. Transport impl can feel free to ignore it.
	Delay bool
}

Options provides additional hints and information for message transmission.

type ServerTransport

type ServerTransport interface {
	// WriteStatus sends the status of a stream to the client.
	WriteStatus(s *Stream, statusCode codes.Code, statusDesc string) error
	// Write sends the data for the given stream.
	Write(s *Stream, data []byte, opts *Options) error
	// WriteHeader sends the header metedata for the given stream.
	WriteHeader(s *Stream, md metadata.MD) error
	// HandleStreams receives incoming streams using the given handler.
	HandleStreams(func(*Stream))
	// 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
}

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

func NewServerTransport

func NewServerTransport(protocol string, conn net.Conn, maxStreams uint32, authInfo credentials.AuthInfo) (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 StreamFromContext

func StreamFromContext(ctx context.Context) (s *Stream, ok bool)

StreamFromContext returns the stream saved in ctx.

func (*Stream) Context

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

Context returns the context of the stream.

func (*Stream) Header

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

Header 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 cancelled/expired.

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 the data available for this Stream from the transport and passes them into the decoder, which converts them into a gRPC message stream. The error is io.EOF when the stream is done or another non-nil error if the stream broke.

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) ServerTransport

func (s *Stream) ServerTransport() ServerTransport

ServerTransport returns the underlying ServerTransport for the stream. The client side stream always returns nil.

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 only be called at most once. Server side only.

func (*Stream) StatusCode

func (s *Stream) StatusCode() codes.Code

StatusCode returns statusCode received from the server.

func (*Stream) StatusDesc

func (s *Stream) StatusDesc() string

StatusDesc returns statusDesc received from the server.

func (*Stream) TraceContext

func (s *Stream) TraceContext(tr trace.Trace)

TraceContext recreates the context of s with a trace.Trace.

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.

type StreamError

type StreamError struct {
	Code codes.Code
	Desc string
}

StreamError is an error that only affects one stream within a connection.

func ContextErr

func ContextErr(err error) StreamError

ContextErr converts the error from context package into a StreamError.

func StreamErrorf

func StreamErrorf(c codes.Code, format string, a ...interface{}) StreamError

StreamErrorf creates an StreamError with the specified error code and description.

func (StreamError) Error

func (e StreamError) Error() string