kafka

Documentation

Index

• type CommonKafka
  • func NewCommonKafka(ctx context.Context, di ConfigDI, topic string) CommonKafka
• type ConfigDI
• type KafkaConfig
  • func (c *KafkaConfig) NewKafkaReader(ctx context.Context, groupID, topic string, l log.Logger) Reader
  • func (c *KafkaConfig) NewKafkaWriter(ctx context.Context, topic string) Writer
• type Reader
• type ReaderHandler
• type Writer

Types

type CommonKafka

type CommonKafka interface {
	GetKafkaWriter() Writer
	Close()
	SetLog(log.Logger)
}

func NewCommonKafka

func NewCommonKafka(ctx context.Context, di ConfigDI, topic string) CommonKafka

type ConfigDI

type ConfigDI interface {
	NewKafkaWriter(ctx context.Context, topic string) Writer
	NewKafkaReader(ctx context.Context, groupID, topic string, l log.Logger) Reader
}

type KafkaConfig

type KafkaConfig struct {
	Brokers []string
}

func (*KafkaConfig) NewKafkaReader

func (c *KafkaConfig) NewKafkaReader(ctx context.Context, groupID, topic string, l log.Logger) Reader

func (*KafkaConfig) NewKafkaWriter

func (c *KafkaConfig) NewKafkaWriter(ctx context.Context, topic string) Writer

type Reader

type Reader interface {
	Read() (map[string]string, []byte, error)
	ReadHandler(handler ReaderHandler) error
	Close() error
}

type ReaderHandler

type ReaderHandler func(headers map[string]string, data []byte) error

type Writer

type Writer interface {
	SetLog(log.Logger)
	Message(headers map[string][]byte, msg []byte) error
	Close() error
}