kafka

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

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

type PublisherAsync 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
}

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.
	OTELEnabled bool
}

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