engine

package
v0.3.1-pre7 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 14, 2023 License: MIT Imports: 12 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrConsumerNotRegister = errors.New("consumer not register")
	ErrProducerNotRegister = errors.New("producer not register")
	ErrPMNotFound          = errors.New("ProducerMessage not found in frame")
)

Functions

This section is empty.

Types

type ChainArgs added in v0.3.2 

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

type Config 

type Config struct {
	Host             string          `json:"host"`
	Port             string          `json:"port"`
	MaxOpenConn      int             `json:"max_open_conn"` // 允许的最大连接数, 即 生产者+消费者最多有 MaxOpenConn 个
	BufferSize       int             `json:"buffer_size"`   // 生产者消息历史记录最大数量
	HeartbeatTimeout float64         `json:"heartbeat_timeout"`
	Logger           logger.Iface    `json:"-"`
	Crypto           proto.Crypto    `json:"-"` // 加密器
	Token            string          `json:"-"` // 注册认证密钥
	EventHandler     EventHandler    `json:"-"` // 事件触发器
	Ctx              context.Context `json:"-"`
	// contains filtered or unexported fields
}

func (*Config) Clean 

func (c *Config) Clean() *Config

type Consumer 

type Consumer struct {
	Addr string          `json:"addr"`
	Conf *ConsumerConfig `json:"conf"`
	Conn transfer.Conn   `json:"-"`
	// contains filtered or unexported fields
}

func (*Consumer) Index added in v0.3.3 

func (c *Consumer) Index() int

func (*Consumer) IsFree added in v0.3.3 

func (c *Consumer) IsFree() bool

func (*Consumer) NeedConfirm 

func (c *Consumer) NeedConfirm() bool

NeedConfirm 是否需要返回确认消息给客户端

func (*Consumer) Send 

func (c *Consumer) Send(msg proto.Message) error

Send 向消费者客户端推送消息, 此操作是线程安全的

func (*Consumer) SetConn added in v0.3.3 

func (c *Consumer) SetConn(r transfer.Conn) *Consumer

type ConsumerConfig 

type ConsumerConfig struct {
	Topics []string      `json:"topics"`
	Ack    proto.AckType `json:"ack"`
}

type DefaultEventHandler added in v0.3.3 

type DefaultEventHandler struct{}

func (DefaultEventHandler) OnCMConsumed added in v0.3.3 

func (e DefaultEventHandler) OnCMConsumed(_ *HistoryRecord)

func (DefaultEventHandler) OnConsumerClosed added in v0.3.3 

func (e DefaultEventHandler) OnConsumerClosed(_ string)

func (DefaultEventHandler) OnConsumerHeartbeatTimeout added in v0.3.3 

func (e DefaultEventHandler) OnConsumerHeartbeatTimeout(event TimeoutEvent)

func (DefaultEventHandler) OnConsumerRegister added in v0.3.3 

func (e DefaultEventHandler) OnConsumerRegister(_ string)

func (DefaultEventHandler) OnConsumerRegisterTimeout added in v0.3.3 

func (e DefaultEventHandler) OnConsumerRegisterTimeout(event TimeoutEvent)

func (DefaultEventHandler) OnFrameParseError added in v0.3.3 

func (e DefaultEventHandler) OnFrameParseError(_ *proto.TransferFrame, _ transfer.Conn)

func (DefaultEventHandler) OnNotImplementMessageType added in v0.3.3 

func (e DefaultEventHandler) OnNotImplementMessageType(frame *proto.TransferFrame, con transfer.Conn) (bool, error)

func (DefaultEventHandler) OnProducerClosed added in v0.3.3 

func (e DefaultEventHandler) OnProducerClosed(_ string)

func (DefaultEventHandler) OnProducerHeartbeatTimeout added in v0.3.3 

func (e DefaultEventHandler) OnProducerHeartbeatTimeout(event TimeoutEvent)

func (DefaultEventHandler) OnProducerRegister added in v0.3.3 

func (e DefaultEventHandler) OnProducerRegister(_ string)

func (DefaultEventHandler) OnProducerRegisterTimeout added in v0.3.3 

func (e DefaultEventHandler) OnProducerRegisterTimeout(event TimeoutEvent)

type Engine 

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

func New 

func New(cs ...Config) *Engine

New 创建一个新的服务器

func (*Engine) AddTopic 

func (e *Engine) AddTopic(name []byte) *Topic

AddTopic 添加一个新的topic,如果topic以存在则跳过

func (*Engine) BindMessageHandler 

func (e *Engine) BindMessageHandler(m proto.Message, handler HookHandler, texts ...string) error

BindMessageHandler 绑定一个自实现的协议处理器, 

参数m为实现了 proto.Message 接口的协议,

参数handler则为收到此协议后的同步处理函数, 如果需要在处理完成之后向客户端返回消息,则直接就地修改frame参数,
	并返回 true 和 nil, 除此之外,则不会向客户端返回任何消息
	HookHandler 的第一个参数为接收到的消息帧,需自行解码, 第二个参数为当前的客户端连接,
	此方法需返回(是否返回数据,处理是否正确)两个参数.

参数texts则为协议m的摘要名称

func (*Engine) Ctx added in v0.3.3 

func (e *Engine) Ctx() context.Context

func (*Engine) EventHandler 

func (e *Engine) EventHandler() EventHandler

func (*Engine) GetTopic 

func (e *Engine) GetTopic(name []byte) *Topic

GetTopic 获取topic,并在不存在时自动新建一个topic

func (*Engine) GetTopicOffset 

func (e *Engine) GetTopicOffset(name []byte) uint64

GetTopicOffset 查询指定topic当前的消息偏移量

func (*Engine) HeartbeatInterval added in v0.3.3 

func (e *Engine) HeartbeatInterval() float64

HeartbeatInterval 心跳周期间隔

func (*Engine) IsTokenCorrect added in v0.3.2 

func (e *Engine) IsTokenCorrect(token string) bool

IsTokenCorrect 判断客户端的token是否正确，若未开启token验证，则始终正确

func (*Engine) Logger 

func (e *Engine) Logger() logger.Iface

func (*Engine) NeedToken added in v0.3.1 

func (e *Engine) NeedToken() bool

NeedToken 是否需要密钥认证

func (*Engine) ProducerSendInterval 

func (e *Engine) ProducerSendInterval() time.Duration

ProducerSendInterval 允许生产者发送数据间隔

func (*Engine) Publisher 

func (e *Engine) Publisher(msg *proto.PMessage) uint64

Publisher 发布消息,并返回此消息在当前topic中的偏移量

func (*Engine) QueryConsumer 

func (e *Engine) QueryConsumer(addr string) (*Consumer, bool)

QueryConsumer 查询消费者记录, 若未注册则返回nil

func (*Engine) QueryProducer 

func (e *Engine) QueryProducer(addr string) (*Producer, bool)

QueryProducer 查询生产者记录, 若未注册则返回nil

func (*Engine) RangeConsumer 

func (e *Engine) RangeConsumer(fn func(c *Consumer) bool)

RangeConsumer if false returned, for-loop will stop

func (*Engine) RangeProducer 

func (e *Engine) RangeProducer(fn func(p *Producer) bool)

RangeProducer if false returned, for-loop will stop

func (*Engine) RangeTopic 

func (e *Engine) RangeTopic(fn func(topic *Topic) bool)

RangeTopic if false returned, for-loop will stop

func (*Engine) RemoveConsumer 

func (e *Engine) RemoveConsumer(addr string)

RemoveConsumer 删除一个消费者

func (*Engine) RemoveProducer 

func (e *Engine) RemoveProducer(addr string)

RemoveProducer 删除一个生产者

func (*Engine) ReplaceTransfer 

func (e *Engine) ReplaceTransfer(transfer transfer.Transfer) *Engine

ReplaceTransfer 替换传输层实现

func (*Engine) Serve added in v0.3.1 

func (e *Engine) Serve() error

Serve 阻塞运行

func (*Engine) SetEventHandler added in v0.3.2 

func (e *Engine) SetEventHandler(handler EventHandler) *Engine

SetEventHandler 设置事件触发器

func (*Engine) SetTopicHistoryBufferSize 

func (e *Engine) SetTopicHistoryBufferSize(size int) *Engine

SetTopicHistoryBufferSize 设置topic历史数据缓存大小, 对于修改前已经创建的topic不受影响 

@param size	int 历史数据缓存大小,[1, 10000)

func (*Engine) Stop added in v0.3.1 

func (e *Engine) Stop()

type EventHandler 

type EventHandler interface {
	// OnFrameParseError 当来自客户端消息帧解析出错时触发的事件(同步调用)
	OnFrameParseError(frame *proto.TransferFrame, con transfer.Conn)
	// OnConsumerRegister 当消费者注册成功时触发的事件(异步调用)
	OnConsumerRegister(addr string)
	// OnProducerRegister 当生产者注册成功时触发的事件(异步调用)
	OnProducerRegister(addr string)
	// OnConsumerClosed 当消费者关闭连接时触发的事件(异步调用)
	OnConsumerClosed(addr string)
	// OnProducerClosed 当生产者关闭连接时触发的事件(异步调用)
	OnProducerClosed(addr string)
	// OnConsumerHeartbeatTimeout 当消费者心跳超时触发的事件(异步调用)
	OnConsumerHeartbeatTimeout(event TimeoutEvent)
	// OnProducerHeartbeatTimeout 当生产者心跳超时时触发的事件(异步调用)
	OnProducerHeartbeatTimeout(event TimeoutEvent)
	// OnConsumerRegisterTimeout 当消费者连接成功后不注册引发的超时事件(异步调用)
	OnConsumerRegisterTimeout(event TimeoutEvent)
	// OnProducerRegisterTimeout 当消生产者连接成功后不注册引发的超时事件(异步调用)
	OnProducerRegisterTimeout(event TimeoutEvent)
	// OnNotImplementMessageType 当收到一个未实现的消息帧时触发的事件(同步调用)
	OnNotImplementMessageType(frame *proto.TransferFrame, con transfer.Conn) (bool, error)
	// OnCMConsumed 当一个消费者被消费成功(成功发送给全部消费者)后时触发的事件(同步调用)
	OnCMConsumed(record *HistoryRecord)
}

EventHandler 事件触发器

type FlowHandler added in v0.3.2 

type FlowHandler func(args *ChainArgs) (stop bool)

type HistoryRecord 

type HistoryRecord struct {
	Topic       []byte            // 历史记录所属的topic
	Offset      []byte            // 历史记录所属的偏移量
	Time        int64             // 历史记录创建时间戳,而非CM被创建的事件戳
	MessageType proto.MessageType // CM协议类型,以此来反序列化
	Stream      []byte            // CM序列化字节流
	// contains filtered or unexported fields
}

type Hook 

type Hook struct {
	Type    proto.MessageType
	Handler HookHandler
	IsAsync bool
}

type HookHandler 

type HookHandler func(frame *proto.TransferFrame, con transfer.Conn) (goon bool, err error)

HookHandler 消息处理方法

type Monitor added in v0.3.3 

type Monitor struct {
	cronjob.Job
	// contains filtered or unexported fields
}

Monitor 监视器 1. 检测连接成功但不注册的客户端 2. 检测心跳超时的客户端 超时时主动断开连接

func (*Monitor) Do added in v0.3.3 

func (k *Monitor) Do(ctx context.Context) error

func (*Monitor) Interval added in v0.3.3 

func (k *Monitor) Interval() time.Duration

func (*Monitor) OnClientClosed added in v0.3.3 

func (k *Monitor) OnClientClosed(addr string)

OnClientClosed 连接关闭，清空时间信息

func (*Monitor) OnClientConnected added in v0.3.3 

func (k *Monitor) OnClientConnected(addr string)

func (*Monitor) OnClientHeartbeat added in v0.3.3 

func (k *Monitor) OnClientHeartbeat(addr string)

func (*Monitor) OnClientRegistered added in v0.3.3 

func (k *Monitor) OnClientRegistered(addr string, linkType proto.LinkType)

func (*Monitor) OnStartup added in v0.3.3 

func (k *Monitor) OnStartup()

func (*Monitor) ReadClientTimeInfo added in v0.3.3 

func (k *Monitor) ReadClientTimeInfo(addr string, linkType proto.LinkType) *TimeInfo

func (*Monitor) String added in v0.3.3 

func (k *Monitor) String() string

type Producer 

type Producer struct {
	Addr string          `json:"addr"`
	Conf *ProducerConfig `json:"conf"`
	Conn transfer.Conn   `json:"-"`
	// contains filtered or unexported fields
}

func (*Producer) Index added in v0.3.3 

func (p *Producer) Index() int

func (*Producer) IsFree added in v0.3.3 

func (p *Producer) IsFree() bool

func (*Producer) NeedConfirm 

func (p *Producer) NeedConfirm() bool

func (*Producer) SetConn added in v0.3.3 

func (p *Producer) SetConn(r transfer.Conn) *Producer

type ProducerConfig 

type ProducerConfig struct {
	Ack proto.AckType `json:"ack"`
	// 定时器间隔，单位ms，仅生产者有效，生产者需要按照此间隔发送帧消息
	TickerInterval time.Duration `json:"ticker_duration"`
}

type TimeInfo added in v0.3.3 

type TimeInfo struct {
	Addr         string         `json:"addr"`
	LinkType     proto.LinkType `json:"link_type"`
	ConnectedAt  int64          `json:"connected_at"`  // 连接成功时间戳
	RegisteredAt int64          `json:"registered_at"` // 注册成功时间戳
	HeartbeatAt  int64          `json:"heartbeat_at"`  // 最近的一个心跳时间戳
}

TimeInfo 关于监视器有关的时间信息

func (*TimeInfo) IsFree added in v0.3.3 

func (t *TimeInfo) IsFree() bool

func (*TimeInfo) IsRegistered added in v0.3.3 

func (t *TimeInfo) IsRegistered() bool

func (*TimeInfo) Reset added in v0.3.3 

func (t *TimeInfo) Reset()

type TimeoutEvent added in v0.3.3 

type TimeoutEvent struct {
	Addr            string           `json:"addr,omitempty"`
	EventType       TimeoutEventType `json:"event_type,omitempty"`
	LinkType        proto.LinkType   `json:"link_type,omitempty"`
	TimeoutInterval float64          `json:"timeout_interval,omitempty"`
	ConnectedAt     int64            `json:"connected_at"`
	TimeoutAt       int64            `json:"timeout_at"`
}

TimeoutEvent 超时事件

type TimeoutEventType added in v0.3.3 

type TimeoutEventType string
const (
	HeartbeatTimeoutEvent TimeoutEventType = "HEARTBEAT_TIMEOUT"
	RegisterTimeoutEvent  TimeoutEventType = "REGISTER_TIMEOUT"
)

type Topic 

type Topic struct {
	Name        []byte `json:"name"`         // 唯一标识
	HistorySize int    `json:"history_size"` // 生产者消息缓冲区大小
	Offset      uint64 `json:"offset"`       // 当前数据偏移量,仅用于模糊显示
	// contains filtered or unexported fields
}

func NewTopic 

func NewTopic(name []byte, bufferSize, historySize int, onConsumed func(record *HistoryRecord)) *Topic

func (*Topic) AddConsumer 

func (t *Topic) AddConsumer(con *Consumer)

AddConsumer 添加一个消费者

func (*Topic) IterConsumer 

func (t *Topic) IterConsumer(fn func(c *Consumer))

IterConsumer 逐个迭代现有消费者

func (*Topic) Publisher 

func (t *Topic) Publisher(pm *proto.PMessage) uint64

Publisher 发布消费者消息,此处会将来自生产者的消息转换成消费者消息

func (*Topic) RemoveConsumer 

func (t *Topic) RemoveConsumer(addr string)

RemoveConsumer 移除一个消费者

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.