kafkax

type BatchConsumerConfig struct {
	// MaxBatchSize max amount of elements the batch will contain.
	// Default value is 100 if nothing is specified.
	MaxBatchSize int16
	// MaxWaitTime max time that it will be waited until MaxBatchSize elements are received.
	// Default value is 100ms if nothing is specified.
	MaxWaitTime time.Duration
}

type ConsumerModel int

type DefaultMarshaler struct{}

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

type Marshaler interface {
	Marshal(topic string, msg *message.Message) (*sarama.ProducerMessage, error)
}

type MarshalerUnmarshaler interface {
	Marshaler
	Unmarshaler
}

type MessageHandler interface {
	ProcessMessages(
		ctx context.Context,
		kafkaMessages <-chan *sarama.ConsumerMessage,
		sess sarama.ConsumerGroupSession,
		messageLogFields watermill.LogFields,
	) error
}

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

type PartitionOffset map[int32]int64

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

type PublisherConfig struct {
	// Kafka brokers list.
	Brokers []string

	// Marshaler is used to marshal messages from Watermill format into Kafka format.
	Marshaler Marshaler

	// OverwriteSaramaConfig holds additional sarama settings.
	OverwriteSaramaConfig *sarama.Config

	// If true then each sent message will be wrapped with Opentelemetry tracing, provided by otelsarama.
	OTELEnabled bool

	// Tracer is used to trace Kafka messages.
	// If nil, then no tracing will be used.
	Tracer SaramaTracer
}

type SaramaTracer interface {
	WrapConsumer(sarama.Consumer, otelsaramax.ConsumerInfo) sarama.Consumer
	// WrapPartitionConsumer(sarama.PartitionConsumer) sarama.PartitionConsumer
	WrapConsumerGroupHandler(sarama.ConsumerGroupHandler, otelsaramax.ConsumerInfo) sarama.ConsumerGroupHandler
	WrapSyncProducer(*sarama.Config, sarama.SyncProducer) sarama.SyncProducer
}

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

type SubscriberConfig struct {
	// Kafka brokers list.
	Brokers []string

	// Unmarshaler is used to unmarshal messages from Kafka format into Watermill format.
	Unmarshaler Unmarshaler

	// OverwriteSaramaConfig holds additional sarama settings.
	OverwriteSaramaConfig *sarama.Config

	// Kafka consumer group.
	// When empty, all messages from all partitions will be returned.
	ConsumerGroup string

	// How long after Nack message should be redelivered.
	NackResendSleep time.Duration

	// How long about unsuccessful reconnecting next reconnect will occur.
	ReconnectRetrySleep time.Duration

	InitializeTopicDetails *sarama.TopicDetail

	// If true then each consumed message will be wrapped with Opentelemetry tracing, provided by otelsarama.
	//
	// Deprecated: pass OTELSaramaTracer to Tracer field instead.
	OTELEnabled bool

	// Tracer is used to trace Kafka messages.
	// If nil, then no tracing will be used.
	Tracer SaramaTracer

	// ConsumerModel indicates which type of consumer should be used
	ConsumerModel ConsumerModel
	// When set to not nil, consumption will be performed in batches and this configuration will be used
	BatchConsumerConfig *BatchConsumerConfig
}

type Unmarshaler interface {
	Unmarshal(*sarama.ConsumerMessage) (*message.Message, error)
}