kafka

Documentation

Index

• func NewKafkaReceiver(address []string, topic string, group string, logger log.Logger, ...) (event.Receiver, func(), error)
• func NewKafkaSender(address []string, topic string, logger log.Logger, cfg *conf.Event_Kafka) (event.Sender, func(), error)
• type OTelInterceptor
  • func NewOTelInterceptor(kind SpanKind, brokers []string) *OTelInterceptor
  • func (oi *OTelInterceptor) EndConsumerSpan(ctx context.Context, span trace.Span, err error)
  • func (oi *OTelInterceptor) OnSend(msg *sarama.ProducerMessage)
  • func (oi *OTelInterceptor) StartConsumerSpan(ctx context.Context, group string, msg *sarama.ConsumerMessage) (context.Context, trace.Span)
• type SpanKind

Functions

func NewKafkaReceiver

func NewKafkaReceiver(address []string, topic string, group string, logger log.Logger, cfg *conf.Event_Kafka) (event.Receiver, func(), error)

func NewKafkaSender

func NewKafkaSender(address []string, topic string, logger log.Logger, cfg *conf.Event_Kafka) (event.Sender, func(), error)

Types

type OTelInterceptor

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

func NewOTelInterceptor

func NewOTelInterceptor(kind SpanKind, brokers []string) *OTelInterceptor

NewOTelInterceptor processes span for intercepted messages and add some headers with the span data.

func (*OTelInterceptor) EndConsumerSpan

func (oi *OTelInterceptor) EndConsumerSpan(ctx context.Context, span trace.Span, err error)

func (*OTelInterceptor) OnSend

func (oi *OTelInterceptor) OnSend(msg *sarama.ProducerMessage)

func (*OTelInterceptor) StartConsumerSpan

func (oi *OTelInterceptor) StartConsumerSpan(ctx context.Context, group string, msg *sarama.ConsumerMessage) (context.Context, trace.Span)

type SpanKind

type SpanKind string

const (
	KindProducer SpanKind = "PRODUCER"
	KindConsumer SpanKind = "CONSUMER"
)