gateway

Documentation

Index

• Variables
• type CallOption
  • func WithNoReply() CallOption
  • func WithNode(nodeID string) CallOption
• type Client
  • func NewClient(dialURL string) (*Client, error)
  • func NewQUICClient(dialURL string, tlsConfig *tls.Config, quicConfig *quic.Config) (*Client, error)
  • func (c *Client) Call(serviceCode int32, method string, arg proto.Message, options ...CallOption) error
  • func (c *Client) OnReceive(serviceCode int32, messageType int32, handler any)
  • func (c *Client) SetDefaultHandler(handler func(reply *nh.Reply))
• type ConnectInterceptor
• type DisconnectInterceptor
• type GoPool
• type Initializer
• type Option
  • func WithConnectInterceptor(interceptor ConnectInterceptor) Option
  • func WithDisconnectInterceptor(interceptor DisconnectInterceptor) Option
  • func WithEventBus(bus *event.Bus) Option
  • func WithGoPool(pool GoPool) Option
  • func WithInitializer(initializer Initializer) Option
  • func WithKeepaliveInterval(interval time.Duration) Option
  • func WithMulticast(multicast multicast.Subscriber) Option
  • func WithRegistry(registry *cluster.Registry) Option
  • func WithRequestDeadline(deadline time.Duration) Option
  • func WithRequestInterceptor(interceptor RequestInterceptor) Option
  • func WithRequestLogger(logger logger.Logger) Option
  • func WithTransporter(transporter Transporter) Option
• type Options
  • func (opt *Options) Validate() error
• type Proxy
  • func NewProxy(nodeID ulid.ULID, opt ...Option) (*Proxy, error)
  • func (p *Proxy) CompleteNodeEntry(entry *cluster.NodeEntry)
  • func (p *Proxy) Name() string
  • func (p *Proxy) NewGRPCService() nh.GatewayServer
  • func (p *Proxy) Start(ctx context.Context) error
  • func (p *Proxy) Stop(ctx context.Context)
• type RequestInterceptor
• type Session
• type Transporter
  • func BindQUICServer(conn net.PacketConn, tlsConfig *tls.Config, quicConfig *quic.Config) Transporter
  • func BindTCPServer(listener net.Listener) Transporter
  • func BindWSServer(listener net.Listener, urlPath string) Transporter
  • func NewQUICServer(listenAddr string, tlsConfig *tls.Config, quicConfig *quic.Config) Transporter
  • func NewTCPServer(listenAddr string) Transporter
  • func NewWSServer(listenAddr string, urlPath string) Transporter

Variables

var (
	// MaxMessageSize 客户端消息最大长度，默认64KB
	MaxMessageSize = 64 * 1024
)

var (
	// Upgrader websocket upgrader
	Upgrader = websocket.Upgrader{
		HandshakeTimeout: 10 * time.Second,
		ReadBufferSize:   1024,
		WriteBufferSize:  1024,
		WriteBufferPool:  &sync.Pool{},
	}
)

Types

type CallOption

type CallOption func(req *nh.Request)

CallOption 调用选项

func WithNoReply

func WithNoReply() CallOption

WithNoReply 不需要回复

func WithNode

func WithNode(nodeID string) CallOption

WithNode 指定节点

type Client

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

Client 网关客户端，用于测试及演示

func NewClient

func NewClient(dialURL string) (*Client, error)

NewClient 创建客户端

func NewQUICClient

func NewQUICClient(dialURL string, tlsConfig *tls.Config, quicConfig *quic.Config) (*Client, error)

NewQUICClient 创建QUIC客户端

func (*Client) Call

func (c *Client) Call(serviceCode int32, method string, arg proto.Message, options ...CallOption) error

Call 发起远程调用

func (*Client) OnReceive

func (c *Client) OnReceive(serviceCode int32, messageType int32, handler any)

OnReceive 注册消息处理器

Example:

client.OnReceive(gateway.ServiceCode, int32(gatewaypb.Protocol_RPC_ERROR), func(requestID uint32, msg *gatewaypb.RPCError) {
	// ...
})

func (*Client) SetDefaultHandler

func (c *Client) SetDefaultHandler(handler func(reply *nh.Reply))

SetDefaultHandler 设置默认消息处理器

type ConnectInterceptor

type ConnectInterceptor func(ctx context.Context, sess Session) error

ConnectInterceptor 在连接创建之后执行自定义操作，返回错误会中断连接

type DisconnectInterceptor

type DisconnectInterceptor func(ctx context.Context, sess Session)

DisconnectInterceptor 在连接断开前执行自定操作

type GoPool

type GoPool interface {
	Submit(task func()) error
}

GoPool goroutine pool

type Initializer

type Initializer func(ctx context.Context, sess Session) (userID string, md metadata.MD, err error)

Initializer 连接初始化，会在客户端与网关建立网络连接后调用，初始化完成之后网关才会开始转发请求

metadata用于网关转发的所有grpc request， userID会作为session唯一标识使用，会被自动加入到metadata中， 如果希望中断初始化且不打印错误日志，return io.EOF错误即可

type Option

type Option func(opt *Options)

Option 网关配置选项

func WithConnectInterceptor

func WithConnectInterceptor(interceptor ConnectInterceptor) Option

WithConnectInterceptor 设置连接拦截器

func WithDisconnectInterceptor

func WithDisconnectInterceptor(interceptor DisconnectInterceptor) Option

WithDisconnectInterceptor 设置断开连接拦截器

func WithEventBus

func WithEventBus(bus *event.Bus) Option

WithEventBus 设置事件总线

func WithGoPool

func WithGoPool(pool GoPool) Option

WithGoPool 设置goroutine pool

func WithInitializer

func WithInitializer(initializer Initializer) Option

WithInitializer 设置连接初始化逻辑

func WithKeepaliveInterval

func WithKeepaliveInterval(interval time.Duration) Option

WithKeepaliveInterval 设置网络连接保持活跃时间，默认1分钟

客户端在没有业务消息的情况下，需要定时向服务器端发送心跳消息 服务器端如果检测到客户端连接超过这个时间还没有任何读写，就会认为此连接已断线，会触发主动断线操作

func WithMulticast

func WithMulticast(multicast multicast.Subscriber) Option

WithMulticast 设置广播组件

func WithRegistry

func WithRegistry(registry *cluster.Registry) Option

WithRegistry 设置服务注册中心

func WithRequestDeadline

func WithRequestDeadline(deadline time.Duration) Option

WithRequestDeadline 设置每个请求的超时时间，默认5秒

func WithRequestInterceptor

func WithRequestInterceptor(interceptor RequestInterceptor) Option

WithRequestInterceptor 设置请求拦截器

func WithRequestLogger

func WithRequestLogger(logger logger.Logger) Option

WithRequestLogger 设置请求日志记录器

func WithTransporter

func WithTransporter(transporter Transporter) Option

WithTransporter 设置传输层

type Options

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

Options 网关配置

func (*Options) Validate

func (opt *Options) Validate() error

Validate 有效性检查

type Proxy

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

Proxy 客户端会话运行环境

func NewProxy

func NewProxy(nodeID ulid.ULID, opt ...Option) (*Proxy, error)

NewProxy 构造函数

func (*Proxy) CompleteNodeEntry

func (p *Proxy) CompleteNodeEntry(entry *cluster.NodeEntry)

CompleteNodeEntry 补全节点信息

func (*Proxy) Name

func (p *Proxy) Name() string

Name 服务名称

func (*Proxy) NewGRPCService

func (p *Proxy) NewGRPCService() nh.GatewayServer

NewGRPCService 网关管理服务

func (*Proxy) Start

func (p *Proxy) Start(ctx context.Context) error

Start 启动服务

func (*Proxy) Stop

func (p *Proxy) Stop(ctx context.Context)

Stop 停止服务

type RequestInterceptor

type RequestInterceptor func(ctx context.Context, sess Session, req *nh.Request) (pass bool, err error)

RequestInterceptor 请求拦截器

每次收到客户端请求都会执行，return pass=false会中断当次请求

type Session

type Session interface {
	ID() string
	SetID(string)
	SetMetadata(metadata.MD)
	MetadataCopy() metadata.MD
	Recv(*nh.Request) error
	Send(*nh.Reply) error
	LocalAddr() string
	RemoteAddr() string
	LastRWTime() time.Time
	Close() error
	LogValue() slog.Value
}

Session 连接会话

type Transporter

type Transporter interface {
	CompleteNodeEntry(entry *cluster.NodeEntry)
	Serve(ctx context.Context) (chan Session, error)
	Shutdown(ctx context.Context) error
}

Transporter 网关传输层接口

func BindQUICServer

func BindQUICServer(conn net.PacketConn, tlsConfig *tls.Config, quicConfig *quic.Config) Transporter

BindQUICServer 绑定QUIC服务器

func BindTCPServer

func BindTCPServer(listener net.Listener) Transporter

BindTCPServer 绑定TCP服务器

func BindWSServer

func BindWSServer(listener net.Listener, urlPath string) Transporter

BindWSServer 绑定websocket服务器

func NewQUICServer

func NewQUICServer(listenAddr string, tlsConfig *tls.Config, quicConfig *quic.Config) Transporter

NewQUICServer 构造函数

func NewTCPServer

func NewTCPServer(listenAddr string) Transporter

NewTCPServer 构造函数

func NewWSServer

func NewWSServer(listenAddr string, urlPath string) Transporter

NewWSServer 构造函数