call

Documentation

Index

• Constants
• Variables
• func InitUnifiedClientConnectionManager(ctx context.Context, logger *slog.Logger)
• func NewStub(name string, reg *codegen.Registration, conn Connection, injectRetries int) codegen.Stub
• func Serve(ctx context.Context, l Listener, opts ServerOptions) error
• type Balancer
  • func RoundRobin() Balancer
• type CallOptions
• type ClientOptions
• type Connection
  • func Connect(ctx context.Context, resolver Resolver, opts ClientOptions) (Connection, error)
• type Endpoint
• type Handler
• type HandlerMap
  • func NewHandlerMap() *HandlerMap
  • func (hm *HandlerMap) AddHandlers(component string, impl any) error
  • func (hm *HandlerMap) Set(component, method string, handler Handler)
• type Listener
• type MethodKey
  • func MakeMethodKey(component, method string) MethodKey
• type NetEndpoint
  • func ParseNetEndpoint(endpoint string) (NetEndpoint, error)
  • func TCP(address string) NetEndpoint
  • func UDP(address string) NetEndpoint
  • func Unix(address string) NetEndpoint
  • func (n NetEndpoint) Address() string
  • func (n NetEndpoint) Dial(ctx context.Context) (net.Conn, error)
• type ReplicaConnection
• type Resolver
  • func NewConstantResolver(endpoints ...Endpoint) Resolver
• type ServerOptions
• type Version

Constants

const (
	CommunicationError transportError = iota
	Unreachable
)

Variables

var Missing = Version{Opaque: "__tombstone__"}

Functions

func InitUnifiedClientConnectionManager

func InitUnifiedClientConnectionManager(ctx context.Context, logger *slog.Logger)

func NewStub

func NewStub(name string, reg *codegen.Registration, conn Connection, injectRetries int) codegen.Stub

func Serve

func Serve(ctx context.Context, l Listener, opts ServerOptions) error

Serve 开始监听连接

Types

type Balancer

type Balancer interface {
	Add(ReplicaConnection)
	Remove(ReplicaConnection)

	Pick(CallOptions) (ReplicaConnection, bool)
}

func RoundRobin

func RoundRobin() Balancer

type CallOptions

type CallOptions struct {
	Retry bool

	ShardKey uint64
}

type ClientOptions

type ClientOptions struct {
	Balancer
	Logger *slog.Logger

	// OptimisticSpinDuration 使用乐观锁自选等待结果的时间
	OptimisticSpinDuration time.Duration

	// 所有小于这个限制的写入数据,在都会被压成一个 buffer 再发送
	// 如果为0，会选一个合适的值， 如果是负值，这不生效
	WriteFlattenLimit int
}

type Connection

type Connection interface {
	Call(context.Context, MethodKey, []byte, CallOptions) ([]byte, error)

	Close()
}

Connection 发送rpc的连接

func Connect

func Connect(ctx context.Context, resolver Resolver, opts ClientOptions) (Connection, error)

type Endpoint

type Endpoint interface {
	Dial(ctx context.Context) (net.Conn, error)

	Address() string
}

type Handler

type Handler func(context.Context, []byte) ([]byte, error)

type HandlerMap

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

func NewHandlerMap

func NewHandlerMap() *HandlerMap

func (*HandlerMap) AddHandlers

func (hm *HandlerMap) AddHandlers(component string, impl any) error

func (*HandlerMap) Set

func (hm *HandlerMap) Set(component, method string, handler Handler)

type Listener

type Listener interface {
	Accept() (net.Conn, *HandlerMap, error)
	Close() error
	Addr() net.Addr
}

type MethodKey

type MethodKey [16]byte

func MakeMethodKey

func MakeMethodKey(component, method string) MethodKey

type NetEndpoint

type NetEndpoint struct {
	Net  string // e.g., "tcp", "udp", "unix"
	Addr string // e.g., "localhost:8000"
}

func ParseNetEndpoint

func ParseNetEndpoint(endpoint string) (NetEndpoint, error)

func TCP

func TCP(address string) NetEndpoint

func UDP

func UDP(address string) NetEndpoint

func Unix

func Unix(address string) NetEndpoint

func (NetEndpoint) Address

func (n NetEndpoint) Address() string

func (NetEndpoint) Dial

func (n NetEndpoint) Dial(ctx context.Context) (net.Conn, error)

type ReplicaConnection

type ReplicaConnection interface {
	Address() string
}

type Resolver

type Resolver interface {
	// IsConstant 返回一个resolver是否不变的
	IsConstant() bool

	Resolve(ctx context.Context, version *Version) ([]Endpoint, *Version, error)
}

func NewConstantResolver

func NewConstantResolver(endpoints ...Endpoint) Resolver

type ServerOptions

type ServerOptions struct {
	Logger *slog.Logger
	Tracer trace.Tracer

	InlineHandlerDuration time.Duration
	WriteFlattenLimit     int
}

type Version

type Version struct {
	Opaque string
}