kafka

Oct 12, 2023

Changes in this version

+ func MessageKeyFromCtx(ctx context.Context) ([]byte, bool)

Jul 7, 2023

Changes in this version

+ type OTELSaramaTracer struct

+ func (t OTELSaramaTracer) WrapConsumer(c sarama.Consumer) sarama.Consumer

+ func (t OTELSaramaTracer) WrapConsumerGroupHandler(h sarama.ConsumerGroupHandler) sarama.ConsumerGroupHandler

+ func (t OTELSaramaTracer) WrapPartitionConsumer(pc sarama.PartitionConsumer) sarama.PartitionConsumer

+ func (t OTELSaramaTracer) WrapSyncProducer(cfg *sarama.Config, p sarama.SyncProducer) sarama.SyncProducer

type PublisherConfig

+ Tracer SaramaTracer

+ type SaramaTracer interface

+ WrapConsumer func(sarama.Consumer) sarama.Consumer

+ WrapConsumerGroupHandler func(sarama.ConsumerGroupHandler) sarama.ConsumerGroupHandler

+ WrapPartitionConsumer func(sarama.PartitionConsumer) sarama.PartitionConsumer

+ WrapSyncProducer func(*sarama.Config, sarama.SyncProducer) sarama.SyncProducer

+ func NewOTELSaramaTracer() SaramaTracer

type SubscriberConfig

+ Tracer SaramaTracer

Mar 31, 2022

Changes in this version

type PublisherConfig

+ OTELEnabled bool

type SubscriberConfig

+ OTELEnabled bool

Dec 14, 2019

Changes in this version

+ func MessageTimestampFromCtx(ctx context.Context) (time.Time, bool)

Dec 9, 2019

Changes in this version

+ func MessagePartitionFromCtx(ctx context.Context) (int32, bool)

+ func MessagePartitionOffsetFromCtx(ctx context.Context) (int64, bool)

+ type PartitionOffset map[int32]int64

type Subscriber

+ func (s *Subscriber) PartitionOffset(topic string) (PartitionOffset, error)

Sep 16, 2019

Changes in this version

+ const NoSleep

+ const UUIDHeaderKey

+ func DefaultSaramaSubscriberConfig() *sarama.Config

+ func DefaultSaramaSyncPublisherConfig() *sarama.Config

+ func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) (message.Publisher, error)

+ func NewSubscriber(config SubscriberConfig, logger watermill.LoggerAdapter) (message.Subscriber, error)

+ type DefaultMarshaler struct

+ func (DefaultMarshaler) Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error)

+ func (DefaultMarshaler) Unmarshal(kafkaMsg *sarama.ConsumerMessage) (*message.Message, error)

+ type GeneratePartitionKey func(topic string, msg *message.Message) (string, error)

+ type Marshaler interface

+ Marshal func(topic string, msg *message.Message) (*sarama.ProducerMessage, error)

+ type MarshalerUnmarshaler interface

+ func NewWithPartitioningMarshaler(generatePartitionKey GeneratePartitionKey) MarshalerUnmarshaler

+ type Publisher struct

+ func (p *Publisher) Close() error

+ func (p *Publisher) Publish(topic string, msgs ...*message.Message) error

+ type PublisherConfig struct

+ Brokers []string

+ Marshaler Marshaler

+ OverwriteSaramaConfig *sarama.Config

+ func (c PublisherConfig) Validate() error

+ type Subscriber struct

+ func (s *Subscriber) Close() error

+ func (s *Subscriber) Subscribe(ctx context.Context, topic string) (<-chan *message.Message, error)

+ func (s *Subscriber) SubscribeInitialize(topic string) (err error)

+ type SubscriberConfig struct

+ Brokers []string

+ ConsumerGroup string

+ InitializeTopicDetails *sarama.TopicDetail

+ NackResendSleep time.Duration

+ OverwriteSaramaConfig *sarama.Config

+ ReconnectRetrySleep time.Duration

+ Unmarshaler Unmarshaler

+ func (c SubscriberConfig) Validate() error

+ type Unmarshaler interface

+ Unmarshal func(*sarama.ConsumerMessage) (*message.Message, error)