engine

Oct 23, 2023

Changes in this version

+ type ConsumerTopic struct

+ Addr string

+ Topics []string

type Engine

+ func (e *Engine) Stat() *Statistic

type HistoryRecord

+ Error string

+ Key []byte

+ Value []byte

+ type HistoryRecordStatus string

+ const BuildFailed

+ type Statistic struct

+ func (k Statistic) ConsumerTopics() []*ConsumerTopic

+ func (k Statistic) LatestRecord() []*HistoryRecord

+ func (k Statistic) Producers() []string

+ func (k Statistic) TopicConsumers() []*TopicConsumer

+ func (k Statistic) TopicsName() []string

+ func (k Statistic) TopicsOffset() []*TopicOffset

type Topic

+ func (t *Topic) LatestMessage() *HistoryRecord

+ func (t *Topic) RangeConsumer(fn func(c *Consumer))

+ type TopicConsumer struct

+ Consumers []string

+ Name string

+ type TopicOffset struct

+ Name string

+ Offset uint64

Aug 21, 2023

Changes in this version

type ChainArgs

+ func (args *ChainArgs) SetError(err error) *ChainArgs

+ func (args *ChainArgs) StopError() error

type Engine

+ func (e *Engine) SetCryptoPlan(option string, key ...string) *Engine

type Topic

+ func (t *Topic) SetCrypto(crypto proto.Crypto) *Topic

+ func (t *Topic) SetOnConsumed(onConsumed func(record *HistoryRecord)) *Topic

Aug 20, 2023

Changes in this version

+ var ErrNoNeedToReply = errors.New("no need to reply to the client")

type ChainArgs

+ func (args *ChainArgs) NoReplyReason() error

+ func (args *ChainArgs) ReplyClient() bool

+ func (args *ChainArgs) Reset()

+ func (args *ChainArgs) SetStopFlag(errs ...error)

type Config

+ Ctx context.Context

+ HeartbeatTimeout float64

type Consumer

+ func (c *Consumer) Index() int

+ func (c *Consumer) IsFree() bool

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

+ type DefaultEventHandler struct

+ func (e DefaultEventHandler) OnCMConsumed(_ *HistoryRecord)

+ func (e DefaultEventHandler) OnConsumerClosed(_ string)

+ func (e DefaultEventHandler) OnConsumerHeartbeatTimeout(event TimeoutEvent)

+ func (e DefaultEventHandler) OnConsumerRegister(_ string)

+ func (e DefaultEventHandler) OnConsumerRegisterTimeout(event TimeoutEvent)

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

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

+ func (e DefaultEventHandler) OnProducerClosed(_ string)

+ func (e DefaultEventHandler) OnProducerHeartbeatTimeout(event TimeoutEvent)

+ func (e DefaultEventHandler) OnProducerRegister(_ string)

+ func (e DefaultEventHandler) OnProducerRegisterTimeout(event TimeoutEvent)

+ type EPool struct

type Engine

+ func (e *Engine) Crypto() proto.Crypto

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

+ func (e *Engine) HeartbeatInterval() float64

+ func (e *Engine) SetCrypto(cry proto.Crypto) *Engine

+ func (e *Engine) TokenCrypto() *proto.TokenCrypto

type EventHandler

+ OnConsumerHeartbeatTimeout func(event TimeoutEvent)

+ OnConsumerRegisterTimeout func(event TimeoutEvent)

+ OnProducerHeartbeatTimeout func(event TimeoutEvent)

+ OnProducerRegisterTimeout func(event TimeoutEvent)

type HistoryRecord

+ Frame *proto.TransferFrame

type Hook

+ ACKDefined bool

+ type Monitor struct

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

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

+ func (k *Monitor) OnClientClosed(addr string)

+ func (k *Monitor) OnClientConnected(addr string)

+ func (k *Monitor) OnClientHeartbeat(addr string)

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

+ func (k *Monitor) OnStartup()

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

+ func (k *Monitor) String() string

type Producer

+ func (p *Producer) Index() int

+ func (p *Producer) IsFree() bool

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

+ type TimeInfo struct

+ Addr string

+ ConnectedAt int64

+ HeartbeatAt int64

+ LinkType proto.LinkType

+ RegisteredAt int64

+ func (t *TimeInfo) IsFree() bool

+ func (t *TimeInfo) IsRegistered() bool

+ func (t *TimeInfo) Reset()

+ type TimeoutEvent struct

+ Addr string

+ ConnectedAt int64

+ EventType TimeoutEventType

+ LinkType proto.LinkType

+ TimeoutAt int64

+ TimeoutInterval float64

+ type TimeoutEventType string

+ const HeartbeatTimeoutEvent

+ const RegisterTimeoutEvent

Aug 6, 2023

Changes in this version

+ type ChainArgs struct

type Engine

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

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

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

Aug 4, 2023

Changes in this version

type Config

+ Token string

type Engine

+ func (e *Engine) NeedToken() bool

+ func (e *Engine) Serve() error

+ func (e *Engine) Stop()

Aug 1, 2023

Changes in this version

+ var ErrConsumerNotRegister = errors.New("consumer not register")

+ var ErrPMNotFound = errors.New("ProducerMessage not found in frame")

+ var ErrProducerNotRegister = errors.New("producer not register")

+ type Config struct

+ BufferSize int

+ Crypto proto.Crypto

+ EventHandler EventHandler

+ Host string

+ Logger logger.Iface

+ MaxOpenConn int

+ Port string

+ func (c *Config) Clean() *Config

+ type Consumer struct

+ Addr string

+ Conf *ConsumerConfig

+ Conn *tcp.Remote

+ func (c *Consumer) NeedConfirm() bool

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

+ type ConsumerConfig struct

+ Ack proto.AckType

+ Topics []string

+ type Engine struct

+ func New(cs ...Config) *Engine

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

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

+ func (e *Engine) EventHandler() EventHandler

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

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

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

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

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

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

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

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

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

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

+ func (e *Engine) RemoveConsumer(addr string)

+ func (e *Engine) RemoveProducer(addr string)

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

+ func (e *Engine) Run()

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

+ type EventHandler interface

+ OnCMConsumed func(record *HistoryRecord)

+ OnConsumerClosed func(addr string)

+ OnConsumerRegister func(addr string)

+ OnFrameParseError func(frame *proto.TransferFrame, r *tcp.Remote)

+ OnNotImplementMessageType func(frame *proto.TransferFrame, r *tcp.Remote) (bool, error)

+ OnProducerClosed func(addr string)

+ OnProducerRegister func(addr string)

+ type HistoryRecord struct

+ MessageType proto.MessageType

+ Offset []byte

+ Stream []byte

+ Time int64

+ Topic []byte

+ type Hook struct

+ Handler HookHandler

+ IsAsync bool

+ Type proto.MessageType

+ type HookHandler func(frame *proto.TransferFrame, r *tcp.Remote) (bool, error)

+ type Producer struct

+ Addr string

+ Conf *ProducerConfig

+ Conn *tcp.Remote

+ func (p *Producer) NeedConfirm() bool

+ type ProducerConfig struct

+ Ack proto.AckType

+ TickerInterval time.Duration

+ type TCPTransfer struct

+ func (t *TCPTransfer) Handler(r *tcp.Remote) error

+ func (t *TCPTransfer) OnAccepted(r *tcp.Remote) error

+ func (t *TCPTransfer) OnClosed(r *tcp.Remote) error

+ func (t *TCPTransfer) Serve() error

+ func (t *TCPTransfer) SetHost(host string)

+ func (t *TCPTransfer) SetLogger(logger logger.Iface)

+ func (t *TCPTransfer) SetMaxOpenConn(num int)

+ func (t *TCPTransfer) SetOnClosedHandler(fn func(addr string))

+ func (t *TCPTransfer) SetOnConnectedHandler(fn func(r *tcp.Remote))

+ func (t *TCPTransfer) SetOnFrameParseErrorHandler(fn func(frame *proto.TransferFrame, r *tcp.Remote))

+ func (t *TCPTransfer) SetOnReceivedHandler(fn func(frame *proto.TransferFrame, r *tcp.Remote))

+ func (t *TCPTransfer) SetPort(port string)

+ func (t *TCPTransfer) Stop()

+ type Topic struct

+ HistorySize int

+ Name []byte

+ Offset uint64

+ func NewTopic(name []byte, bufferSize, historySize int, ...) *Topic

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

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

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

+ func (t *Topic) RemoveConsumer(addr string)

+ type Transfer interface

+ Serve func() error

+ SetHost func(host string)

+ SetLogger func(logger logger.Iface)

+ SetMaxOpenConn func(num int)

+ SetOnClosedHandler func(fn func(addr string))

+ SetOnConnectedHandler func(fn func(r *tcp.Remote))

+ SetOnFrameParseErrorHandler func(fn func(frame *proto.TransferFrame, r *tcp.Remote))

+ SetOnReceivedHandler func(fn func(frame *proto.TransferFrame, r *tcp.Remote))

+ SetPort func(port string)

+ Stop func()