kakafka

Documentation

Index

• type Consumer
  • func NewConsumer(ctx context.Context, log zerolog.Logger, cfg config.KafkaCfg) *Consumer
  • func (c *Consumer) Run() error
• type Producer
  • func NewProducer(ctx context.Context, cfg config.KafkaCfg, logger zerolog.Logger) *Producer
  • func (p *Producer) Write(ctx context.Context, data []byte)

Types

type Consumer

type Consumer struct {
	Messages <-chan kafka.Message
	Commiter func(ctx context.Context, msgs ...kafka.Message) error
	// contains filtered or unexported fields
}

func NewConsumer

func NewConsumer(ctx context.Context, log zerolog.Logger, cfg config.KafkaCfg) *Consumer

func (*Consumer) Run

func (c *Consumer) Run() error

type Producer

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

func NewProducer

func NewProducer(ctx context.Context, cfg config.KafkaCfg, logger zerolog.Logger) *Producer

func (*Producer) Write

func (p *Producer) Write(ctx context.Context, data []byte)