kafka

Documentation

Index

• type Composed
  • func Start(ctx context.Context, options Options) (Composed, error)
  • func (c *Composed) Terminate(ctx context.Context)
• type Consumer
  • 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
  • func (options *ConsumerOptions) StartConsumer(ctx context.Context) (Consumer, *sync.WaitGroup, error)
• type Container
  • func (c *Container) Terminate(ctx context.Context)
• type Options
• type Producer
  • func (p *Producer) Close() error
  • func (p *Producer) Send(topic string, pkey string, entry sarama.Encoder)
• type ProducerOptions
  • func (options *ProducerOptions) NewProducer() (*Producer, error)

Types

type Composed

type Composed struct {
	Kafka     *Container
	Zookeeper *tczk.Container
	Network   testcontainers.Network
}

Composed ...

func Start

func Start(ctx context.Context, options Options) (Composed, error)

Start ...

func (*Composed) Terminate

func (c *Composed) Terminate(ctx context.Context)

Terminate ...

type Consumer

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

Consumer ...

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 ...

func (*ConsumerOptions) StartConsumer

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

StartConsumer ...

type Container

type Container struct {
	tc.ContainerConfig
	Container testcontainers.Container
	Host      string
	Port      int
	Brokers   []string
	Listeners []string
	Version   string
}

Container ...

func (*Container) Terminate

func (c *Container) Terminate(ctx context.Context)

Terminate ...

type Options

type Options struct {
	tc.ContainerOptions
	LogLevel          string
	KafkaImageTag     string
	ZookeeperImageTag string
}

Options ...

type Producer

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

Producer ...

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 ...

func (*ProducerOptions) NewProducer

func (options *ProducerOptions) NewProducer() (*Producer, error)

NewProducer ...