kafka

Documentation

Index

• type AsyncProducer
  • func NewAsyncProducer(brokers []string, oo ...OptionFunc) (*AsyncProducer, error)
  • func (ap *AsyncProducer) Close() error
  • func (ap *AsyncProducer) Error() <-chan error
  • func (ap *AsyncProducer) Send(ctx context.Context, msg *Message) error
• type Message
  • func NewJSONMessage(t string, d interface{}) (*Message, error)
  • func NewJSONMessageWithKey(t string, d interface{}, k string) (*Message, error)
  • func NewMessage(t string, b []byte) *Message
  • func NewMessageWithKey(t string, b []byte, k string) (*Message, error)
• type OptionFunc
  • func RequiredAcksPolicy(ack RequiredAcks) OptionFunc
  • func Timeouts(dial time.Duration) OptionFunc
  • func Version(version string) OptionFunc
• type Producer
• type RequiredAcks

Types

type AsyncProducer

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

AsyncProducer defines a async Kafka producer.

func NewAsyncProducer

func NewAsyncProducer(brokers []string, oo ...OptionFunc) (*AsyncProducer, error)

NewAsyncProducer creates a new async producer with default configuration.

func (*AsyncProducer) Close

func (ap *AsyncProducer) Close() error

Close gracefully the producer.

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 NewJSONMessage

func NewJSONMessage(t string, d interface{}) (*Message, error)

NewJSONMessage creates a new message with a JSON encoded body.

func NewJSONMessageWithKey

func NewJSONMessageWithKey(t string, d interface{}, k string) (*Message, error)

NewJSONMessageWithKey creates a new message with a JSON encoded body and a message key

func NewMessage

func NewMessage(t string, b []byte) *Message

NewMessage creates a new message.

func NewMessageWithKey

func NewMessageWithKey(t string, b []byte, k string) (*Message, error)

type OptionFunc

type OptionFunc func(*AsyncProducer) error

OptionFunc definition for configuring the async producer in a functional way.

func RequiredAcksPolicy

func RequiredAcksPolicy(ack RequiredAcks) OptionFunc

RequiredAcksPolicy option for adjusting how many replica acknowledgements broker must see before responding.

func Timeouts

func Timeouts(dial time.Duration) OptionFunc

Timeouts option for setting the timeouts.

func Version

func Version(version string) OptionFunc

Version option for setting the version.

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
)