kafka

Documentation

Index

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

Types

type AsyncBuilder

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

AsyncBuilder gathers all required and optional properties, in order to construct a Kafka AsyncProducer.

func NewBuilder

func NewBuilder(brokers []string) *AsyncBuilder

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

func (*AsyncBuilder) Create

func (ab *AsyncBuilder) Create() (*AsyncProducer, error)

Create constructs the AsyncProducer component by applying the gathered properties.

func (*AsyncBuilder) WithEncoder

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

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

func (*AsyncBuilder) WithRequiredAcksPolicy

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

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

func (*AsyncBuilder) WithTimeout

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

WithTimeout sets the dial timeout for the AsyncProducer.

func (*AsyncBuilder) WithVersion

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

WithVersion sets the kafka versionfor the AsyncProducer.

type AsyncProducer

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

AsyncProducer defines a async Kafka producer.

func (*AsyncProducer) ActiveBrokers

func (ap *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) Error

func (ap *AsyncProducer) Error() <-chan error

Error returns a chanel to monitor for errors.

func (*AsyncProducer) Send

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

Send a message to a topic.

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
	Error() <-chan 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
)