kafka

Documentation

Index

• Variables
• func CreateConfig(cfg *config.Config) common.Config
• type ConsumerGroup
• type Kafka
  • func NewKafka(cfg common.Config, logger *zap.Logger) *Kafka
  • func (k *Kafka) NewConsumerGroup(nameGroup string, opts ...consumergroup.ConsumerGroupOption) (ConsumerGroup, error)
  • func (k *Kafka) NewSyncProducer(opts ...syncproducer.SyncProducerOption) (SyncProducer, error)
• type SyncProducer

Variables

var Module = types.Module{
	{CreateFunc: NewKafka},
	{CreateFunc: CreateConfig},
}

Functions

func CreateConfig

func CreateConfig(cfg *config.Config) common.Config

Types

type ConsumerGroup

type ConsumerGroup interface {
	AddHandler(topic common.Topic, hm common.MessageHandler)
	Consume()
	Close() error
}

type Kafka

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

func NewKafka

func NewKafka(cfg common.Config, logger *zap.Logger) *Kafka

func (*Kafka) NewConsumerGroup

func (k *Kafka) NewConsumerGroup(nameGroup string, opts ...consumergroup.ConsumerGroupOption) (ConsumerGroup, error)

func (*Kafka) NewSyncProducer

func (k *Kafka) NewSyncProducer(opts ...syncproducer.SyncProducerOption) (SyncProducer, error)

type SyncProducer

type SyncProducer interface {
	Send(topic string, message []byte, opts ...syncproducer.SendOptions) error
	Close() error
}