kafka

Documentation

Index

• type AsyncProducer
  • func (p *AsyncProducer) ActiveBrokers() []string
  • func (ap *AsyncProducer) Close() error
  • func (ap *AsyncProducer) Send(ctx context.Context, msg *Message) error
• type Builder
  • func NewBuilder(brokers []string) *Builder
  • func (ab *Builder) CreateAsync() (*AsyncProducer, <-chan error, error)
  • func (ab *Builder) CreateSync() (*SyncProducer, error)
  • func (ab *Builder) WithEncoder(enc encoding.EncodeFunc, contentType string) *Builder
  • func (ab *Builder) WithRequiredAcksPolicy(ack RequiredAcks) *Builder
  • func (ab *Builder) WithTimeout(dial time.Duration) *Builder
  • func (ab *Builder) WithVersion(version string) *Builder
• type Message
  • func NewMessage(t string, b interface{}) *Message
  • func NewMessageWithKey(t string, b interface{}, k string) (*Message, error)
• type Producer
• type RequiredAcks
• type SyncProducer
  • func (p *SyncProducer) ActiveBrokers() []string
  • func (p *SyncProducer) Close() error
  • func (p *SyncProducer) Send(ctx context.Context, msg *Message) error

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 *Message) 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 gathers all required and optional properties, in order to construct a Kafka AsyncProducer/SyncProducer.

func NewBuilder

func NewBuilder(brokers []string) *Builder

NewBuilder initiates the AsyncProducer/SyncProducer builder chain. The builder instantiates the component using default values for EncodeFunc and Content-Type header.

func (*Builder) CreateAsync

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

CreateAsync constructs the AsyncProducer component by applying the gathered properties.

func (*Builder) CreateSync

func (ab *Builder) CreateSync() (*SyncProducer, error)

CreateSync constructs the SyncProducer component by applying the gathered properties.

func (*Builder) WithEncoder

func (ab *Builder) WithEncoder(enc encoding.EncodeFunc, contentType string) *Builder

WithEncoder sets a specific encoder implementation and Content-Type string header; if no option is provided it defaults to json.

func (*Builder) WithRequiredAcksPolicy

func (ab *Builder) WithRequiredAcksPolicy(ack RequiredAcks) *Builder

WithRequiredAcksPolicy adjusts how many replica acknowledgements broker must see before responding.

func (*Builder) WithTimeout

func (ab *Builder) WithTimeout(dial time.Duration) *Builder

WithTimeout sets the dial timeout for the sync or async producer.

func (*Builder) WithVersion

func (ab *Builder) WithVersion(version string) *Builder

WithVersion sets the kafka versionfor the AsyncProducer/SyncProducer.

type Message

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

Message abstraction of a Kafka message.

func NewMessage

func NewMessage(t string, b interface{}) *Message

NewMessage creates a new message.

func NewMessageWithKey

func NewMessageWithKey(t string, b interface{}, k string) (*Message, error)

NewMessageWithKey creates a new message with an associated key.

type Producer

type Producer interface {
	Send(ctx context.Context, msg *Message) error
	Close() error
}

Producer interface for Kafka.

type RequiredAcks

type RequiredAcks int16

RequiredAcks is used in Produce Requests to tell the broker how many replica acknowledgements it must see before responding.

const (
	// NoResponse doesn't send any response, the TCP ACK is all you get.
	NoResponse RequiredAcks = 0
	// WaitForLocal waits for only the local commit to succeed before responding.
	WaitForLocal RequiredAcks = 1
	// WaitForAll waits for all in-sync replicas to commit before responding.
	WaitForAll RequiredAcks = -1
)

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 *Message) error

Send a message to a topic.