conn

Documentation

Index

• Variables
• func Dial(opts *DialOptions) (net.Conn, error)
• type ConnectionPool
  • func (p *ConnectionPool) Close() error
  • func (p *ConnectionPool) Get(ctx context.Context) (*PoolConn, error)
  • func (p *ConnectionPool) RegisterChecker(interval time.Duration, checker func(*PoolConn) bool)
• type Decoder
• type DialOptions
• type Framer
• type FramerBuilder
• type GetOption
  • func WithDialTLS(certFile, keyFile, caFile, serverName string) GetOption
  • func WithFramerBuilder(fb FramerBuilder) GetOption
• type GetOptions
• type MetaData
• type Msg
• type Option
  • func WithDialTimeout(t time.Duration) Option
  • func WithForceClose(f bool) Option
  • func WithIdleTimeout(t time.Duration) Option
  • func WithMaxActive(s int) Option
  • func WithMaxConnLifetime(t time.Duration) Option
  • func WithMaxIdle(m int) Option
  • func WithMinIdle(n int) Option
  • func WithWait(w bool) Option
• type Options
• type Pool
  • func NewConnectionPool(opt ...Option) Pool
• type PoolConn
  • func (pc *PoolConn) Close() error
  • func (pc *PoolConn) Read(b []byte) (int, error)
  • func (pc *PoolConn) ReadFrame() ([]byte, error)
  • func (pc *PoolConn) Write(b []byte) (int, error)
• type TransportResponseFrame

Variables

var (
	ErrPoolLimit  = errors.New("connection pool limit")  // ErrPoolLimit 连接数量超过限制错误
	ErrPoolClosed = errors.New("connection pool closed") // ErrPoolClosed 连接池关闭错误
	ErrConnClosed = errors.New("conn closed")            // ErrConnClosed 连接关闭
)

var DefaultConnectionPool = NewConnectionPool()

Functions

func Dial

func Dial(opts *DialOptions) (net.Conn, error)

Dial 发起请求

Types

type ConnectionPool

type ConnectionPool struct {
	Dial            func(context.Context) (net.Conn, error) // 初始化连接
	MinIdle         int                                     // 初始连接数
	MaxIdle         int                                     // 最大闲置连接数，0代表不做限制
	MaxActive       int                                     // 最大活跃连接数量，0代表不做限制
	IdleTimeout     time.Duration                           // 空闲连接超时时间
	Wait            bool                                    // 活跃连接达到最大数量时，是否等待
	MaxConnLifetime time.Duration                           // 连接的最大生命周期
	// contains filtered or unexported fields
}

func (*ConnectionPool) Close

func (p *ConnectionPool) Close() error

Close 释放连接

func (*ConnectionPool) Get

func (p *ConnectionPool) Get(ctx context.Context) (*PoolConn, error)

Get 从连接池中获取连接

func (*ConnectionPool) RegisterChecker

func (p *ConnectionPool) RegisterChecker(interval time.Duration, checker func(*PoolConn) bool)

RegisterChecker 注册空闲连接检查方法

type Decoder

type Decoder interface {
	// Decode 解析出帧头，包头，包体
	Decode() (TransportResponseFrame, error)
	UpdateMsg(interface{}, Msg) error
}

Decoder 解码回包

type DialOptions

type DialOptions struct {
	Network       string
	Address       string
	Timeout       time.Duration
	CACertFile    string
	TLSCertFile   string
	TLSKeyFile    string
	TLSServerName string
}

DialOptions 请求参数

type Framer

type Framer interface {
	ReadFrame() ([]byte, error)
}

Framer 读写数据帧

type FramerBuilder

type FramerBuilder interface {
	New(io.Reader) Framer
}

FramerBuilder 通常每个连接Build一个Framer

type GetOption

type GetOption func(*GetOptions)

GetOption Options helper

func WithDialTLS

func WithDialTLS(certFile, keyFile, caFile, serverName string) GetOption

WithDialTLS 设置client支持TLS

func WithFramerBuilder

func WithFramerBuilder(fb FramerBuilder) GetOption

type GetOptions

type GetOptions struct {
	FramerBuilder FramerBuilder
	CACertFile    string // ca证书
	TLSCertFile   string // client证书
	TLSKeyFile    string // client秘钥
	TLSServerName string // client校验server的服务名，不填时默认为http的hostname
}

GetOptions get conn configuration

type MetaData

type MetaData map[string][]byte

MetaData 请求message透传字段信息

type Msg

type Msg interface {
	Context() context.Context
	WithRemoteAddr(addr net.Addr) // server transport 设置上游地址，client设置下游地址
	WithLocalAddr(addr net.Addr)  // server transport 设置本地地址
	RemoteAddr() net.Addr
	LocalAddr() net.Addr
	WithNamespace(string) // server 的 namespace
	Namespace() string

	WithEnvName(string) // server 的环境
	EnvName() string

	WithSetName(string) // server所在的set
	SetName() string

	WithEnvTransfer(string) // 服务透传的环境信息
	EnvTransfer() string

	WithRequestTimeout(time.Duration) // server codec 设置上游超时，client设置下游超时
	RequestTimeout() time.Duration

	WithSerializationType(int)
	SerializationType() int

	WithCompressType(int)
	CompressType() int

	WithServerRPCName(string) // server codec 设置当前server handler调用方法名
	WithClientRPCName(string) // client stub 设置上游调用方法名

	ServerRPCName() string // 当前server handler调用方法名: /grpc.app.server.service/method
	ClientRPCName() string // 调用下游的接口方法名

	WithCallerServiceName(string) // 调用方服务名
	WithCalleeServiceName(string) // 被调用服务名

	WithCallerApp(string) // 主调app server角度是上游的app， client角度是自身的app
	WithCallerServer(string)
	WithCallerService(string)
	WithCallerMethod(string)

	WithCalleeApp(string) // 被调app server角度是自身app，client角度是下游的app
	WithCalleeServer(string)
	WithCalleeService(string)
	WithCalleeMethod(string)

	CallerServiceName() string // 主调服务名: trpc.app.server.service server角度是上游服务名, client角度是自身的服务名
	CallerApp() string         // 主调app server角度是上游的app， client角度是自身的app
	CallerServer() string
	CallerService() string
	CallerMethod() string

	CalleeServiceName() string // 被调服务名 server角度是自身服务名，client角度是下游的服务名
	CalleeApp() string         // 被调app server角度是自身app, client角度是下游的app
	CalleeServer() string
	CalleeService() string
	CalleeMethod() string

	CalleeContainerName() string // 被调服务容器名
	WithCalleeContainerName(string)

	WithServerMetaData(MetaData)
	ServerMetaData() MetaData
	///.....
	WithRequestID(uint32)
	RequestID() uint32
	WithStreamID(uint32)
	StreamID() uint32

	StreamFrame() interface{}
	WithStreamFrame(interface{})
}

Msg 多协议通用的消息数据，业务协议打解包的时候需要设置message内部消息

type Option

type Option func(*Options)

Option Options helper

func WithDialTimeout

func WithDialTimeout(t time.Duration) Option

func WithForceClose

func WithForceClose(f bool) Option

func WithIdleTimeout

func WithIdleTimeout(t time.Duration) Option

func WithMaxActive

func WithMaxActive(s int) Option

func WithMaxConnLifetime

func WithMaxConnLifetime(t time.Duration) Option

func WithMaxIdle

func WithMaxIdle(m int) Option

func WithMinIdle

func WithMinIdle(n int) Option

WithMinIdle 指定最小空闲连接数

func WithWait

func WithWait(w bool) Option

type Options

type Options struct {
	MinIdle         int           // 最低闲置连接数量，为下次io做好准备
	MaxIdle         int           // 最大闲置连接数量，0代表不做闲置
	MaxActive       int           // 最大活跃连接数量，0代表不做限制
	Wait            bool          // 活跃连接达到最大数量时，是否等待
	IdleTimeout     time.Duration // 空闲连接超时时间
	MaxConnLifetime time.Duration // 连接的最大生命周期
	DialTimeout     time.Duration // 建立连接超时时间
	ForceClose      bool
}

Option indicate pool configuration

type Pool

type Pool interface {
	Get(network string, address string, timeout time.Duration, opt ...GetOption) (net.Conn, error)
}

Pool client connection pool

func NewConnectionPool

func NewConnectionPool(opt ...Option) Pool

type PoolConn

type PoolConn struct {
	net.Conn
	// contains filtered or unexported fields
}

PoolConn 连接池中的连接

func (*PoolConn) Close

func (pc *PoolConn) Close() error

Close 重写net.Conn的Close方法，放回连接池

func (*PoolConn) Read

func (pc *PoolConn) Read(b []byte) (int, error)

Read 连接上读取数据

func (*PoolConn) ReadFrame

func (pc *PoolConn) ReadFrame() ([]byte, error)

ReadFrame 读取帧

func (*PoolConn) Write

func (pc *PoolConn) Write(b []byte) (int, error)

Write 连接上发送数据

type TransportResponseFrame

type TransportResponseFrame interface {
	GetRequestID() uint32
	GetResponseBuf() []byte
}

TransportResponseFrame Decode 解析出的回包应该实现接口