kafka

Documentation

Index

• func NewConsumer(address []string, topic string, group string, cfg *event.Config_Kafka) (event.Consumer, error)
• type Consumer
  • func (k *Consumer) Close() error
  • func (k *Consumer) GetTracer() trace.Tracer
  • func (k *Consumer) Process(ctx context.Context, handler event.ConsumerHandler) error
  • func (k *Consumer) ReportSpanAttr() []attribute.KeyValue
• type Producer
  • func NewProducer(address []string, topic string, cfg *event.Config_Kafka) (*Producer, error)
  • func (s *Producer) BatchSend(ctx context.Context, message []event.Event) error
  • func (s *Producer) Close() error
  • func (s *Producer) GetTracer() trace.Tracer
  • func (s *Producer) ReportSpanAttr() []attribute.KeyValue
  • func (s *Producer) Send(ctx context.Context, message event.Event) error

Functions

func NewConsumer

func NewConsumer(address []string, topic string, group string, cfg *event.Config_Kafka) (event.Consumer, error)

Types

type Consumer

type Consumer struct {
	sarama.ConsumerGroup
	// contains filtered or unexported fields
}

func (*Consumer) Close

func (k *Consumer) Close() error

func (*Consumer) GetTracer

func (k *Consumer) GetTracer() trace.Tracer

func (*Consumer) Process

func (k *Consumer) Process(ctx context.Context, handler event.ConsumerHandler) error

func (*Consumer) ReportSpanAttr

func (k *Consumer) ReportSpanAttr() []attribute.KeyValue

type Producer

type Producer struct {
	sarama.SyncProducer
	// contains filtered or unexported fields
}

func NewProducer

func NewProducer(address []string, topic string, cfg *event.Config_Kafka) (*Producer, error)

func (*Producer) BatchSend

func (s *Producer) BatchSend(ctx context.Context, message []event.Event) error

func (*Producer) Close

func (s *Producer) Close() error

func (*Producer) GetTracer

func (s *Producer) GetTracer() trace.Tracer

func (*Producer) ReportSpanAttr

func (s *Producer) ReportSpanAttr() []attribute.KeyValue

func (*Producer) Send

func (s *Producer) Send(ctx context.Context, message event.Event) error