kafka

Documentation

Index

• func ParseBalanceStrategy(assignor string) sarama.BalanceStrategy
• func ParseCodec(codec string) sarama.CompressionCodec
• func ParseRequiredAcks(acks string) sarama.RequiredAcks
• func ParseVersion(config *sarama.Config, version string) error
• func SASLVersion(kafkaVersion sarama.KafkaVersion, saslVersion *int) (int16, error)
• type AsyncProducerResult
• type OptionFn
  • func WithHeader(k, v string) OptionFn
  • func WithKey(key string) OptionFn
• type Options
  • func (o *Options) Fulfil(msg *sarama.ProducerMessage)
• type Producer
  • func (p Producer) Publish(topic string, vv []byte, optionsFns ...OptionFn) (*PublishResponse, error)
• type ProducerConfig
  • func (c *ProducerConfig) NewProducer() (*Producer, error)
• type PublishResponse
• type SyncProducerResult
• type TlsConfig
  • func (tc TlsConfig) Create() *tls.Config

Functions

func ParseBalanceStrategy

func ParseBalanceStrategy(assignor string) sarama.BalanceStrategy

func ParseCodec

func ParseCodec(codec string) sarama.CompressionCodec

func ParseRequiredAcks

func ParseRequiredAcks(acks string) sarama.RequiredAcks

func ParseVersion

func ParseVersion(config *sarama.Config, version string) error

func SASLVersion

func SASLVersion(kafkaVersion sarama.KafkaVersion, saslVersion *int) (int16, error)

Types

type AsyncProducerResult

type AsyncProducerResult struct {
	Enqueued    bool
	ContextDone bool
}

type OptionFn

type OptionFn func(*Options)

func WithHeader

func WithHeader(k, v string) OptionFn

func WithKey

func WithKey(key string) OptionFn

type Options

type Options struct {
	MessageKey string
	Headers    map[string]string
}

func (*Options) Fulfil

func (o *Options) Fulfil(msg *sarama.ProducerMessage)

type Producer

type Producer struct {
	io.Closer
	Config *ProducerConfig
	// contains filtered or unexported fields
}

func (Producer) Publish

func (p Producer) Publish(topic string, vv []byte, optionsFns ...OptionFn) (*PublishResponse, error)

type ProducerConfig

type ProducerConfig struct {
	Topic           string
	Version         string
	Brokers         []string
	Codec           string
	MaxMessageBytes int
	Sync            bool
	RequiredAcks    sarama.RequiredAcks

	TlsConfig TlsConfig
	Context   context.Context

	SASLUser     string
	SASLPassword string
	SASLVersion  *int
}

func (*ProducerConfig) NewProducer

func (c *ProducerConfig) NewProducer() (*Producer, error)

type PublishResponse

type PublishResponse struct {
	Result interface{}
	Topic  string
}

type SyncProducerResult

type SyncProducerResult struct {
	Partition int32
	Offset    int64
}

type TlsConfig

type TlsConfig struct {
	CaFile, CertFile, KeyFile string
	InsecureSkipVerify        bool
}

func (TlsConfig) Create

func (tc TlsConfig) Create() *tls.Config