rpc

Documentation

Index

• Constants
• func ConvertErrorStreamClientInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, ...) (grpc.ClientStream, error)
• func ConvertErrorStreamServerInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, ...) error
• func ConvertErrorUnaryClientInterceptor(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, ...) error
• func ConvertErrorUnaryServerInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, ...) (any, error)
• func ExecuteWithRetry(f func() (any, error), initBackoff float64, maxBackoff float64, ...) (any, error)
• func Listen(netAddr dfnet.NetAddr) (net.Listener, error)
• func ListenWithPortRange(listen string, startPort, endPort int) (net.Listener, int, error)
• func RefresherStreamClientInterceptor(r Refresher) grpc.StreamClientInterceptor
• func RefresherUnaryClientInterceptor(r Refresher) grpc.UnaryClientInterceptor
• func VsockDialer(_ctx context.Context, address string) (net.Conn, error)
• func VsockDialerOption(addrs []dfnet.NetAddr, opts []grpc.DialOption) ([]grpc.DialOption, error)
• type Closer
• type ConnOption
  • func WithConnExpireTime(duration time.Duration) ConnOption
  • func WithDialOption(opts []grpc.DialOption) ConnOption
  • func WithDialTimeout(dialTimeout time.Duration) ConnOption
  • func WithGcConnInterval(gcConnInterval time.Duration) ConnOption
  • func WithGcConnTimeout(gcConnTimeout time.Duration) ConnOption
• type ConnStatus
• type Connection
  • func NewConnection(ctx context.Context, name string, addrs []dfnet.NetAddr, connOpts []ConnOption) *Connection
  • func (conn *Connection) AddServerNodes(addrs []dfnet.NetAddr) error
  • func (conn *Connection) Close() error
  • func (conn *Connection) CorrectKey2NodeRelation(tmpHashKey, realHashKey string)
  • func (conn *Connection) GetClientConn(hashKey string, stick bool) (clientConn *grpc.ClientConn, err error)
  • func (conn *Connection) GetClientConnByTarget(node string) (*grpc.ClientConn, error)
  • func (conn *Connection) GetServerNode(hashKey string) (string, bool)
  • func (conn *Connection) GetState() []dfnet.NetAddr
  • func (conn *Connection) TryMigrate(key string, cause error, exclusiveNodes []string) (preNode string, err error)
  • func (conn *Connection) UpdateAccessNodeMapByHashKey(key string)
  • func (conn *Connection) UpdateAccessNodeMapByServerNode(serverNode string)
  • func (conn *Connection) UpdateState(addrs []dfnet.NetAddr)
• type RateLimiterInterceptor
  • func NewRateLimiterInterceptor(qps float64, burst int64) *RateLimiterInterceptor
  • func (r *RateLimiterInterceptor) Limit() bool
• type Refresher
• type RetryMeta

Constants

const (
	// Identifier of message transmitted or received.
	RPCMessageIDKey = attribute.Key("message.id")

	// The uncompressed size of the message transmitted or received in
	// bytes.
	RPCMessageUncompressedSizeKey = attribute.Key("message.uncompressed_size")
)

Functions

func ConvertErrorStreamClientInterceptor

func ConvertErrorStreamClientInterceptor(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error)

ConvertErrorStreamClientInterceptor returns a new stream client interceptor that convert error when trigger custom error.

func ConvertErrorStreamServerInterceptor

func ConvertErrorStreamServerInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error

ConvertErrorStreamServerInterceptor returns a new stream server interceptor that convert error when trigger custom error.

func ConvertErrorUnaryClientInterceptor

func ConvertErrorUnaryClientInterceptor(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error

ConvertErrorUnaryClientInterceptor returns a new unary client interceptor that convert error when trigger custom error.

func ConvertErrorUnaryServerInterceptor

func ConvertErrorUnaryServerInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error)

ConvertErrorUnaryServerInterceptor returns a new unary server interceptor that convert error when trigger custom error.

func ExecuteWithRetry

func ExecuteWithRetry(f func() (any, error), initBackoff float64, maxBackoff float64, maxAttempts int, cause error) (any, error)

func Listen

func Listen(netAddr dfnet.NetAddr) (net.Listener, error)

Listen wraps net.Listen with dfnet.NetAddr Example:

Listen(dfnet.NetAddr{Type: dfnet.UNIX, Addr: "/var/run/df.sock"})
Listen(dfnet.NetAddr{Type: dfnet.TCP, Addr: ":12345"})

func ListenWithPortRange

func ListenWithPortRange(listen string, startPort, endPort int) (net.Listener, int, error)

ListenWithPortRange tries to listen a port between startPort and endPort, return net.Listener and listen port Example:

ListenWithPortRange("0.0.0.0", 12345, 23456)
ListenWithPortRange("192.168.0.1", 12345, 23456)
ListenWithPortRange("192.168.0.1", 0, 0) // random port

func RefresherStreamClientInterceptor

func RefresherStreamClientInterceptor(r Refresher) grpc.StreamClientInterceptor

StreamClientInterceptor returns a new stream client interceptor that refresh dynconfig addresses when calling error.

func RefresherUnaryClientInterceptor

func RefresherUnaryClientInterceptor(r Refresher) grpc.UnaryClientInterceptor

UnaryClientInterceptor returns a new unary client interceptor that refresh dynconfig addresses when calling error.

func VsockDialer

func VsockDialer(_ctx context.Context, address string) (net.Conn, error)

VsockDialer is the dialer for vsock, it expects `address` to be in dfnet.NetAddr.GetEndpoint() format, that is "vsock://cid:port"

func VsockDialerOption

func VsockDialerOption(addrs []dfnet.NetAddr, opts []grpc.DialOption) ([]grpc.DialOption, error)

If `addrs` are all vsock addresses, add rpc.VsockDialer to DialOption, and return error if addrs have mixed vsock and other connection types.

Types

type Closer

type Closer interface {
	Close() error
}

type ConnOption

type ConnOption interface {
	// contains filtered or unexported methods
}

func WithConnExpireTime

func WithConnExpireTime(duration time.Duration) ConnOption

func WithDialOption

func WithDialOption(opts []grpc.DialOption) ConnOption

func WithDialTimeout

func WithDialTimeout(dialTimeout time.Duration) ConnOption

func WithGcConnInterval

func WithGcConnInterval(gcConnInterval time.Duration) ConnOption

func WithGcConnTimeout

func WithGcConnTimeout(gcConnTimeout time.Duration) ConnOption

type ConnStatus

type ConnStatus string

TODO Perfect state

type Connection

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

func NewConnection

func NewConnection(ctx context.Context, name string, addrs []dfnet.NetAddr, connOpts []ConnOption) *Connection

func (*Connection) AddServerNodes

func (conn *Connection) AddServerNodes(addrs []dfnet.NetAddr) error

func (*Connection) Close

func (conn *Connection) Close() error

func (*Connection) CorrectKey2NodeRelation

func (conn *Connection) CorrectKey2NodeRelation(tmpHashKey, realHashKey string)

func (*Connection) GetClientConn

func (conn *Connection) GetClientConn(hashKey string, stick bool) (clientConn *grpc.ClientConn, err error)

GetClientConn get conn or bind hashKey to candidate node, don't do the migrate action stick whether hash key need already associated with specify node

func (*Connection) GetClientConnByTarget

func (conn *Connection) GetClientConnByTarget(node string) (*grpc.ClientConn, error)

func (*Connection) GetServerNode

func (conn *Connection) GetServerNode(hashKey string) (string, bool)

GetServerNode

func (*Connection) GetState

func (conn *Connection) GetState() []dfnet.NetAddr

func (*Connection) TryMigrate

func (conn *Connection) TryMigrate(key string, cause error, exclusiveNodes []string) (preNode string, err error)

TryMigrate migrate key to another hash node other than exclusiveNodes preNode node before the migration

func (*Connection) UpdateAccessNodeMapByHashKey

func (conn *Connection) UpdateAccessNodeMapByHashKey(key string)

func (*Connection) UpdateAccessNodeMapByServerNode

func (conn *Connection) UpdateAccessNodeMapByServerNode(serverNode string)

func (*Connection) UpdateState

func (conn *Connection) UpdateState(addrs []dfnet.NetAddr)

type RateLimiterInterceptor

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

RateLimiterInterceptor is the interface for ratelimit interceptor.

func NewRateLimiterInterceptor

func NewRateLimiterInterceptor(qps float64, burst int64) *RateLimiterInterceptor

NewRateLimiterInterceptor returns a RateLimiterInterceptor instance.

func (*RateLimiterInterceptor) Limit

func (r *RateLimiterInterceptor) Limit() bool

Limit is the predicate which limits the requests.

type Refresher

type Refresher interface {
	Refresh() error
}

Refresher is the interface for refreshing dynconfig.

type RetryMeta

type RetryMeta struct {
	StreamTimes int     // times of replacing stream on the current client
	MaxAttempts int     // limit times for execute
	InitBackoff float64 // second
	MaxBackOff  float64 // second
}