kafka

Documentation

Index

• type Consumer
  • func ConsumeGroup(ctx context.Context, options ConsumerOptions) (*Consumer, *sync.WaitGroup, error)
  • func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error
  • func (consumer *Consumer) Close() error
  • func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error
  • func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error
• type ConsumerOptions
• type ContainerConnectionConfig
  • func StartKafkaContainer(ctx context.Context, options ContainerOptions) (testcontainers.Container, *ContainerConnectionConfig, testcontainers.Container, ...)
  • func StartStandaloneKafkaContainer(ctx context.Context, options ContainerOptions) (testcontainers.Container, *ContainerConnectionConfig, error)
• type ContainerOptions
• type Producer
  • func CreateProducer(ctx context.Context, options ProducerOptions) (*Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) Send(topic string, pkey string, entry sarama.Encoder)
• type ProducerOptions

Types

type Consumer

type Consumer struct {
	Messages chan *sarama.ConsumerMessage
	// contains filtered or unexported fields
}

Consumer ...

func ConsumeGroup

func ConsumeGroup(ctx context.Context, options ConsumerOptions) (*Consumer, *sync.WaitGroup, error)

ConsumeGroup ...

func (*Consumer) Cleanup

func (consumer *Consumer) Cleanup(sarama.ConsumerGroupSession) error

Cleanup ...

func (*Consumer) Close

func (consumer *Consumer) Close() error

Close ...

func (*Consumer) ConsumeClaim

func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error

ConsumeClaim ...

func (*Consumer) Setup

func (consumer *Consumer) Setup(sarama.ConsumerGroupSession) error

Setup ...

type ConsumerOptions

type ConsumerOptions struct {
	Brokers []string
	Group   string
	Version string
	Topics  []string
}

ConsumerOptions ...

type ContainerConnectionConfig

type ContainerConnectionConfig struct {
	tc.ContainerConfig
	Brokers      []string
	KafkaVersion string
}

ContainerConnectionConfig ...

func StartKafkaContainer

func StartKafkaContainer(ctx context.Context, options ContainerOptions) (testcontainers.Container, *ContainerConnectionConfig, testcontainers.Container, testcontainers.Network, error)

StartKafkaContainer ...

func StartStandaloneKafkaContainer

func StartStandaloneKafkaContainer(ctx context.Context, options ContainerOptions) (testcontainers.Container, *ContainerConnectionConfig, error)

StartStandaloneKafkaContainer ...

type ContainerOptions

type ContainerOptions struct {
	tc.ContainerOptions
	StartZookeeper bool
}

ContainerOptions ...

type Producer

type Producer struct {
	DataCollector     sarama.SyncProducer
	AccessLogProducer sarama.AsyncProducer
}

Producer ...

func CreateProducer

func CreateProducer(ctx context.Context, options ProducerOptions) (*Producer, error)

CreateProducer ...

func (*Producer) Close

func (p *Producer) Close() error

Close ...

func (*Producer) Send

func (p *Producer) Send(topic string, pkey string, entry sarama.Encoder)

Send ...

type ProducerOptions

type ProducerOptions struct {
	Brokers []string
	Group   string
	Version string
	Topics  []string
}

ProducerOptions ...