gateway

Documentation

Index

• Variables
• type Authorizer
• 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 Option
  • func WithAuthorizer(authorizer Authorizer) Option
  • func WithConnectInterceptor(interceptor ConnectInterceptor) Option
  • func WithDisconnectInterceptor(interceptor DisconnectInterceptor) Option
  • func WithEventBus(bus *event.Bus) Option
  • func WithMulticast(multicast multicast.Subscriber) Option
  • func WithRegistry(registry *cluster.Registry) Option
  • func WithRequestInterceptor(interceptor RequestInterceptor) Option
  • func WithRequestLogger(logger logger.Logger) Option
  • func WithTransporter(transporter Transporter) Option
• 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) error
• type RequestInterceptor
• type Session
• type Transporter
  • func NewQUICServer(listenAddr string, tlsConfig *tls.Config, quicConfig *quic.Config) Transporter
  • func NewTCPServer(listenAddr string) Transporter
  • func NewWSServer(listenAddr string) Transporter

Variables

var (
	// DefaultHeartbeatDuration 心跳消息发送时间间隔，连接在超过这个时间没有收发消息，就会发送心跳消息
	DefaultHeartbeatDuration = 1 * time.Minute

	// DefaultHeartbeatTimeout 心跳消息超时时间，默认为心跳消息发送时间间隔的3倍
	DefaultHeartbeatTimeout = 3 * DefaultHeartbeatDuration

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

var ErrDenyByAuthorizer = errors.New("deny by authorizer")

ErrDenyByAuthorizer 身份验证未通过

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

Types

type Authorizer

type Authorizer func(ctx context.Context, sess Session) (userID string, md metadata.MD, ok bool)

Authorizer 身份认证

返回的metadata会在此连接的所有grpc request中携带 返回的userID如果不为空，则会作为会话唯一标识使用，另外也会被自动加入到metadata中 如果返回ok为false，会直接关闭连接 因此如果验证不通过之类的错误，需要在这个函数里面自行处理

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 Option

type Option func(p *Proxy)

Option 网关配置选项

func WithAuthorizer

func WithAuthorizer(authorizer Authorizer) Option

WithAuthorizer 设置身份验证

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 WithMulticast

func WithMulticast(multicast multicast.Subscriber) Option

WithMulticast 设置广播组件

func WithRegistry

func WithRegistry(registry *cluster.Registry) Option

WithRegistry 设置服务注册中心

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

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

NewWSServer 构造函数