agent

Documentation

Index

• func Exit()
• func Init(cfg *Config, log mq.Logger, redis Redis, queueName string, removeCert bool) error
• func Publish(topic string, header map[string]string, msg []byte, opts ...mq.PublishOption) error
• func Subscribe(group string, h Handler, topics []string) error
• func SubscribeWithStrategyOfRetry(group string, h Handler, topics []string, retryNum int) error
• func SubscribeWithStrategyOfSendBack(group string, h Handler, topics []string) error
• type Config
  • func (cfg *Config) Validate() error
• type Handler
• type Redis

Functions

func Exit

func Exit()

func Init

func Init(cfg *Config, log mq.Logger, redis Redis, queueName string, removeCert bool) error

func Publish

func Publish(topic string, header map[string]string, msg []byte, opts ...mq.PublishOption) error

func Subscribe

func Subscribe(group string, h Handler, topics []string) error

func SubscribeWithStrategyOfRetry

func SubscribeWithStrategyOfRetry(group string, h Handler, topics []string, retryNum int) error

func SubscribeWithStrategyOfSendBack

func SubscribeWithStrategyOfSendBack(group string, h Handler, topics []string) error

Types

type Config

type Config struct {
	Address        string `json:"address" required:"true"`
	Version        string `json:"version"` // e.g 2.1.0
	MQCert         string `json:"mq_cert"`
	OTEL           bool   `json:"otel"` // Whether otel tracing is enabled
	Username       string `json:"user_name"`
	Password       string `json:"password"`
	Algorithm      string `json:"algorithm"`
	SkipCertVerify bool   `json:"skip_cert_verify"`
}

func (*Config) Validate

func (cfg *Config) Validate() error

type Handler

type Handler func([]byte, map[string]string) error

Handler

type Redis

type Redis interface {
	RPush(string, interface{}) error
	LPop(string, interface{}) error
	IsKeyNotExists(error) bool
}