listener

Documentation

Index

• func NewConsumerGroupHandler(kafkaListener KafkaListener) sarama.ConsumerGroupHandler
• func ParseBalanceStrategy(partitionAssignor string) (sarama.BalanceStrategy, error)
• func ParseInitialOffset(initialOffset string) (int64, error)
• type Config
  • func (c *Config) ToSaramaConfig() (*sarama.Config, error)
• type KafkaContainer
  • func NewKafkaContainer(kafkaListeners []KafkaListener, listenerConfig *Config) (KafkaContainer, error)
• type KafkaListener

Functions

func NewConsumerGroupHandler

func NewConsumerGroupHandler(kafkaListener KafkaListener) sarama.ConsumerGroupHandler

func ParseBalanceStrategy

func ParseBalanceStrategy(partitionAssignor string) (sarama.BalanceStrategy, error)

func ParseInitialOffset

func ParseInitialOffset(initialOffset string) (int64, error)

Types

type Config

type Config struct {
	// Kafka version.
	Version string

	// Supported values: earliest, latest.
	AutoOffsetReset string

	BootstrapServers []string

	// Supported values: range, sticky, round-robin.
	PartitionAssignmentStrategy string
}

func (*Config) ToSaramaConfig

func (c *Config) ToSaramaConfig() (*sarama.Config, error)

type KafkaContainer

type KafkaContainer interface {
	Run(ctx context.Context) error
}

func NewKafkaContainer

func NewKafkaContainer(kafkaListeners []KafkaListener, listenerConfig *Config) (KafkaContainer, error)

type KafkaListener

type KafkaListener interface {
	Id() string

	GroupId() string

	// Supported value: manual.
	AckMode() string

	Topics() []string

	OnEventFunc() func(record *sarama.ConsumerMessage) error

	// Required if AckMode equals to "manual".
	OnEventAckFunc() func(record *sarama.ConsumerMessage, session sarama.ConsumerGroupSession)
}