kafka

Documentation

Index

• Variables
• func AddTypedHandler[T any](cg ConsumerGroup, topic common.Topic, f common.MessageHandlerTyped[T]) error
• func CreateConfig(cfg *config.Config) common.Config
• func TypedSend[T any](sp SyncProducer, topic string, message T, opts ...syncproducer.SendOptions) error
• 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 AddTypedHandler

func AddTypedHandler[T any](
	cg ConsumerGroup,
	topic common.Topic,
	f common.MessageHandlerTyped[T],
) error

func CreateConfig

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

func TypedSend

func TypedSend[T any](
	sp SyncProducer,
	topic string,
	message T,
	opts ...syncproducer.SendOptions,
) error

Types

type ConsumerGroup

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

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
}