producer

Documentation

Index

• type Adapter
• type Config
• type Producer
  • func New(initConfig *Config) (*Producer, error)
  • func (p *Producer) Close() chan error
  • func (p *Producer) CreateKeyMessage(topic string, key string, value string) *sarama.ProducerMessage
  • func (p *Producer) CreateMessage(topic string, value string) *sarama.ProducerMessage
  • func (p *Producer) EnableLogging()
  • func (p *Producer) Get() *Adapter
  • func (p *Producer) Input() (chan<- *sarama.ProducerMessage, error)
  • func (p *Producer) IsClosed() bool

Types

type Adapter

type Adapter interface {
	AsyncClose()
	Close() error
	Input() chan<- *sarama.ProducerMessage
	Successes() <-chan *sarama.ProducerMessage
	Errors() <-chan *sarama.ProducerError
}

Adapter is the Kafka-Producer interface

type Config

type Config struct {
	ErrHandler   func(*sarama.ProducerError)
	KafkaBrokers []string
	// Allow overwriting default sarama-config
	SaramaConfig *sarama.Config
}

Config wraps configuration for producer

type Producer

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

Producer wraps sarama's producer

func New

func New(initConfig *Config) (*Producer, error)

New returns a configured sarama Kafka-AsyncProducer instance

func (*Producer) Close

func (p *Producer) Close() chan error

Close attempts to close the producer, and returns any occurring errors over channel

func (*Producer) CreateKeyMessage

func (p *Producer) CreateKeyMessage(topic string, key string, value string) *sarama.ProducerMessage

CreateKeyMessage creates producer-formatted message with key

func (*Producer) CreateMessage

func (p *Producer) CreateMessage(topic string, value string) *sarama.ProducerMessage

CreateMessage creates keyless producer-formatted message

func (*Producer) EnableLogging

func (p *Producer) EnableLogging()

EnableLogging logs events to console

func (*Producer) Get

func (p *Producer) Get() *Adapter

Get returns the original Sarama Kafka producer

func (*Producer) Input

func (p *Producer) Input() (chan<- *sarama.ProducerMessage, error)

Input takes Kafka messages to be produced

func (*Producer) IsClosed

func (p *Producer) IsClosed() bool

IsClosed returns a bool specifying if Kafka producer is closed