kafkax

Jun 11, 2024

Changes in this version

type MessageHandler

+ Cleanup func(*sarama.ConsumerGroupSession) error

+ Setup func(*sarama.ConsumerGroupSession) error

Mar 5, 2024

Changes in this version

type Publisher

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

Jan 31, 2024

Changes in this version

+ const NoSleep

+ const UUIDHeaderKey

+ func DefaultSaramaSubscriberConfig() *sarama.Config

+ func DefaultSaramaSyncPublisherConfig() *sarama.Config

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

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

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

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

+ type BatchConsumerConfig struct

+ MaxBatchSize int16

+ MaxWaitTime time.Duration

+ type ConsumerModel int

+ const Batch

+ const Default

+ const PartitionConcurrent

+ 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 MessageHandler interface

+ ProcessMessages func(ctx context.Context, kafkaMessages <-chan *sarama.ConsumerMessage, ...) error

+ func NewBatchedMessageHandler(outputChannel chan<- *message.Message, unmarshaler Unmarshaler, ...) MessageHandler

+ func NewMessageHandler(outputChannel chan<- *message.Message, unmarshaler Unmarshaler, ...) MessageHandler

+ func NewPartitionConcurrentMessageHandler(outputChannel chan<- *message.Message, unmarshaler Unmarshaler, ...) MessageHandler

+ 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 PartitionOffset map[int32]int64

+ type Publisher struct

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

+ func (p *Publisher) Close() error

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

+ type PublisherConfig struct

+ Brokers []string

+ Marshaler Marshaler

+ OTELEnabled bool

+ OverwriteSaramaConfig *sarama.Config

+ Tracer SaramaTracer

+ func (c PublisherConfig) Validate() error

+ 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 Subscriber struct

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

+ func (s *Subscriber) Close() error

+ func (s *Subscriber) PartitionOffset(topic string) (PartitionOffset, 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

+ BatchConsumerConfig *BatchConsumerConfig

+ Brokers []string

+ ConsumerGroup string

+ ConsumerModel ConsumerModel

+ InitializeTopicDetails *sarama.TopicDetail

+ NackResendSleep time.Duration

+ OTELEnabled bool

+ OverwriteSaramaConfig *sarama.Config

+ ReconnectRetrySleep time.Duration

+ Tracer SaramaTracer

+ Unmarshaler Unmarshaler

+ func (c SubscriberConfig) Validate() error

+ type Unmarshaler interface

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