kafka

Documentation

Index

• Constants
• func NewConsumer(conf *Config, topics []string, group string, fn mq.HandleFunc, ...) (mq.Consumer, error)
• type Config
• type Consumer
  • func (c *Consumer) Start() error
  • func (c *Consumer) Stop() error
• type Producer
  • func NewProducer(conf *Config) (*Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) SendMessage(ctx context.Context, topic, key, tag string, msg proto.Message) error

Constants

const (
	TopicCommentPublish      = "comment_publish"
	TopicCommentCacheRebuild = "comment_cache_rebuild"
	TopicCommentOperator     = "comment_operator"

	TopicRelationFollow       = "relation_follow"
	TopicRelationCacheRebuild = "relation_cache_rebuild"
	TopicRelationSyncCount    = "relation_sync_count"
	TopicRelationOperator     = "relation_operator"

	TopicOpusOperator = "opus_operator"
)

Functions

func NewConsumer

func NewConsumer(conf *Config, topics []string, group string, fn mq.HandleFunc, opts ...mq.Option) (mq.Consumer, error)

Types

type Config

type Config struct {
	Addr []string `yaml:"addr"`
}

type Consumer

type Consumer struct {
	*mq.ConsumerMeta
	// contains filtered or unexported fields
}

func (*Consumer) Start

func (c *Consumer) Start() error

func (*Consumer) Stop

func (c *Consumer) Stop() error

type Producer

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

func NewProducer

func NewProducer(conf *Config) (*Producer, error)

func (*Producer) Close

func (p *Producer) Close() error

func (*Producer) SendMessage

func (p *Producer) SendMessage(ctx context.Context, topic, key, tag string, msg proto.Message) error