manager

Documentation

Index

• func HandleCrash()
• type AggregatorManager
  • func (x *AggregatorManager) Aggregator(ctx context.Context, ruleName string) (rabbit.Aggregator, error)
• type FilterManager
  • func (x *FilterManager) Filter(ctx context.Context, ruleName string) (rabbit.Filter, error)
• type Manager
  • func New(name string, opts *Options) (*Manager, error)
  • func (m *Manager) Aggregator(ctx context.Context, ruleName string) (rabbit.Aggregator, error)
  • func (m *Manager) Filter(ctx context.Context, ruleName string) (rabbit.Filter, error)
  • func (m *Manager) Name() string
  • func (m *Manager) RegisterReceivers(receivers ...rabbit.Receiver)
  • func (m *Manager) RuleGroup(ctx context.Context, name string) (*rabbit.RuleGroup, error)
  • func (m *Manager) Sender(ctx context.Context, ruleName string) (rabbit.Sender, error)
  • func (m *Manager) Start(ctx context.Context) error
  • func (m *Manager) Templater(ctx context.Context, ruleName string) (rabbit.Templater, error)
• type Options
• type SenderManager
  • func (x *SenderManager) Sender(ctx context.Context, ruleName string) (rabbit.Sender, error)
• type TemplaterManager
  • func (x *TemplaterManager) Templater(ctx context.Context, ruleName string) (rabbit.Templater, error)

Functions

func HandleCrash

func HandleCrash()

HandleCrash handles panic and recover

Types

type AggregatorManager

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

func (*AggregatorManager) Aggregator

func (x *AggregatorManager) Aggregator(ctx context.Context, ruleName string) (rabbit.Aggregator, error)

type FilterManager

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

func (*FilterManager) Filter

func (x *FilterManager) Filter(ctx context.Context, ruleName string) (rabbit.Filter, error)

type Manager

type Manager struct {
	MaxConcurrentWorkers int
	Started              bool
	Queue                rabbit.MessageQueue

	Log *logr.Logger
	// contains filtered or unexported fields
}

Manager 是一个消息队列的 Manager

func New

func New(name string, opts *Options) (*Manager, error)

New 创建一个 Manager

func (*Manager) Aggregator

func (m *Manager) Aggregator(ctx context.Context, ruleName string) (rabbit.Aggregator, error)

func (*Manager) Filter

func (m *Manager) Filter(ctx context.Context, ruleName string) (rabbit.Filter, error)

func (*Manager) Name

func (m *Manager) Name() string

Name 返回 Manager 的名称

func (*Manager) RegisterReceivers

func (m *Manager) RegisterReceivers(receivers ...rabbit.Receiver)

RegisterReceivers 注册一个或多个 Receiver

func (*Manager) RuleGroup

func (m *Manager) RuleGroup(ctx context.Context, name string) (*rabbit.RuleGroup, error)

func (*Manager) Sender

func (m *Manager) Sender(ctx context.Context, ruleName string) (rabbit.Sender, error)

func (*Manager) Start

func (m *Manager) Start(ctx context.Context) error

Start 启动 Manager

func (*Manager) Templater

func (m *Manager) Templater(ctx context.Context, ruleName string) (rabbit.Templater, error)

type Options

type Options struct {
	// MaxConcurrentWorkers 是最大并发 worker 数量，默认为1
	MaxConcurrentWorkers int

	// Log 日志记录器
	Log *logr.Logger

	// Backoff 消息再次加入时，用来决定消息重新入队的时间
	Backoff *rabbit.Backoff

	// MaxRetries 表示数据最多尝试的入队次数，超过这个次数，数据将被丢弃
	MaxRetries int
}

Options 是 Manager 的配置项

type SenderManager

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

func (*SenderManager) Sender

func (x *SenderManager) Sender(ctx context.Context, ruleName string) (rabbit.Sender, error)

type TemplaterManager

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

func (*TemplaterManager) Templater

func (x *TemplaterManager) Templater(ctx context.Context, ruleName string) (rabbit.Templater, error)