channel

Documentation

Overview

Package channel 提供基于 Channel 的消息分发机制

Index

• Variables
• type Channel
  • func NewChannel(name string, msgType message.MessageType, bufferSize int) *Channel
  • func (c *Channel) AddHandler(h Handler)
  • func (c *Channel) Chan() <-chan *message.Envelope
  • func (c *Channel) Close()
  • func (c *Channel) Name() string
  • func (c *Channel) Process(envelope *message.Envelope) error
  • func (c *Channel) Receive(ctx context.Context) (*message.Envelope, error)
  • func (c *Channel) Send(envelope *message.Envelope) error
  • func (c *Channel) Type() message.MessageType
• type Handler
• type Router
  • func NewRouter(manager *transport.Manager) *Router
  • func (r *Router) GetChannel(msgType message.MessageType) *Channel
  • func (r *Router) GetChannelByName(name string) *Channel
  • func (r *Router) RegisterChannel(ch *Channel)
  • func (r *Router) RegisterTypeChannel(msgType message.MessageType, bufSize int) *Channel
  • func (r *Router) Request(ctx context.Context, envelope *message.Envelope) (*message.Envelope, error)
  • func (r *Router) Send(ctx context.Context, envelope *message.Envelope) error
  • func (r *Router) SetDefaultChannel(ch *Channel)
  • func (r *Router) Start(ctx context.Context) error
  • func (r *Router) Stop()

Variables

var (
	ErrChannelNotFound = errors.New("channel not found")
	ErrChannelClosed   = errors.New("channel closed")
	ErrChannelFull     = errors.New("channel buffer full")
)

常见错误

Types

type Channel

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

Channel 消息通道

func NewChannel

func NewChannel(name string, msgType message.MessageType, bufferSize int) *Channel

NewChannel 创建消息通道

func (*Channel) AddHandler

func (c *Channel) AddHandler(h Handler)

AddHandler 添加消息处理器

func (*Channel) Chan

func (c *Channel) Chan() <-chan *message.Envelope

Chan 获取底层 channel

func (*Channel) Close

func (c *Channel) Close()

Close 关闭通道

func (*Channel) Name

func (c *Channel) Name() string

Name 获取通道名称

func (*Channel) Process

func (c *Channel) Process(envelope *message.Envelope) error

Process 处理消息

func (*Channel) Receive

func (c *Channel) Receive(ctx context.Context) (*message.Envelope, error)

Receive 接收消息

func (*Channel) Send

func (c *Channel) Send(envelope *message.Envelope) error

Send 发送消息到通道

func (*Channel) Type

func (c *Channel) Type() message.MessageType

Type 获取消息类型

type Handler

type Handler func(envelope *message.Envelope) error

Handler 消息处理函数

type Router

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

Router 消息路由器

func NewRouter

func NewRouter(manager *transport.Manager) *Router

NewRouter 创建消息路由器

func (*Router) GetChannel

func (r *Router) GetChannel(msgType message.MessageType) *Channel

GetChannel 获取消息通道

func (*Router) GetChannelByName

func (r *Router) GetChannelByName(name string) *Channel

GetChannelByName 按名称获取通道

func (*Router) RegisterChannel

func (r *Router) RegisterChannel(ch *Channel)

RegisterChannel 注册消息通道

func (*Router) RegisterTypeChannel

func (r *Router) RegisterTypeChannel(msgType message.MessageType, bufSize int) *Channel

RegisterTypeChannel 按类型注册通道

func (*Router) Request

func (r *Router) Request(ctx context.Context, envelope *message.Envelope) (*message.Envelope, error)

Request 发送请求并等待响应

func (*Router) Send

func (r *Router) Send(ctx context.Context, envelope *message.Envelope) error

Send 通过路由器发送消息

func (*Router) SetDefaultChannel

func (r *Router) SetDefaultChannel(ch *Channel)

SetDefaultChannel 设置默认通道

func (*Router) Start

func (r *Router) Start(ctx context.Context) error

Start 启动路由器

func (*Router) Stop

func (r *Router) Stop()

Stop 停止路由器