kafka

Documentation

Overview

Package kafka provides a client with included tracing capabilities.

Index

• func DefaultProducerSaramaConfig(name string, idempotent bool) (*sarama.Config, error)
• type AsyncProducer
  • func (p *AsyncProducer) ActiveBrokers() []string
  • func (ap *AsyncProducer) Close() error
  • func (ap *AsyncProducer) Send(ctx context.Context, msg *sarama.ProducerMessage) error
• type Builder
  • func New(brokers []string, saramaConfig *sarama.Config) *Builder
  • func (b *Builder) Create() (*SyncProducer, error)
  • func (b *Builder) CreateAsync() (*AsyncProducer, <-chan error, error)
• type SyncProducer
  • func (p *SyncProducer) ActiveBrokers() []string
  • func (p *SyncProducer) Close() error
  • func (p *SyncProducer) Send(ctx context.Context, msg *sarama.ProducerMessage) (int32, int64, error)
  • func (p *SyncProducer) SendBatch(ctx context.Context, messages []*sarama.ProducerMessage) error

Functions

func DefaultProducerSaramaConfig

func DefaultProducerSaramaConfig(name string, idempotent bool) (*sarama.Config, error)

DefaultProducerSaramaConfig creates a default Sarama configuration with idempotency enabled. See also: * https://pkg.go.dev/github.com/Shopify/sarama#RequiredAcks * https://pkg.go.dev/github.com/Shopify/sarama#Config

Types

type AsyncProducer

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

AsyncProducer is an asynchronous Kafka producer.

func (*AsyncProducer) ActiveBrokers

func (p *AsyncProducer) ActiveBrokers() []string

ActiveBrokers returns a list of active brokers' addresses.

func (*AsyncProducer) Close

func (ap *AsyncProducer) Close() error

Close shuts down the producer and waits for any buffered messages to be flushed. You must call this function before a producer object passes out of scope, as it may otherwise leak memory.

func (*AsyncProducer) Send

func (ap *AsyncProducer) Send(ctx context.Context, msg *sarama.ProducerMessage) error

Send a message to a topic, asynchronously. Producer errors are queued on the channel obtained during the AsyncProducer creation.

type Builder

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

Builder definition for creating sync and async producers.

func New

func New(brokers []string, saramaConfig *sarama.Config) *Builder

New initiates the AsyncProducer/SyncProducer builder chain with the specified Sarama configuration.

func (*Builder) Create

func (b *Builder) Create() (*SyncProducer, error)

Create a new synchronous producer.

func (*Builder) CreateAsync

func (b *Builder) CreateAsync() (*AsyncProducer, <-chan error, error)

CreateAsync a new asynchronous producer.

type SyncProducer

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

SyncProducer is a synchronous Kafka producer.

func (*SyncProducer) ActiveBrokers

func (p *SyncProducer) ActiveBrokers() []string

ActiveBrokers returns a list of active brokers' addresses.

func (*SyncProducer) Close

func (p *SyncProducer) Close() error

Close shuts down the producer and waits for any buffered messages to be flushed. You must call this function before a producer object passes out of scope, as it may otherwise leak memory.

func (*SyncProducer) Send

func (p *SyncProducer) Send(ctx context.Context, msg *sarama.ProducerMessage) (int32, int64, error)

Send a message to a topic.

func (*SyncProducer) SendBatch

func (p *SyncProducer) SendBatch(ctx context.Context, messages []*sarama.ProducerMessage) error

SendBatch sends a batch to a topic.