kafka

Documentation

Overview

Package kafka contains an implementation of Producer interface that can be used to produce (that is send) messages to properly configured Kafka broker.

Index

• type Producer
  • func New(config *conf.ConfigStruct) (*Producer, error)
  • func (producer *Producer) Close() error
  • func (producer *Producer) ProduceMessage(msg types.ProducerMessage) (partitionID int32, offset int64, err error)
• type SCRAMClient
  • func (x *SCRAMClient) Begin(userName, password, authzID string) (err error)
  • func (x *SCRAMClient) Done() bool
  • func (x *SCRAMClient) Step(challenge string) (response string, err error)

Types

type Producer

type Producer struct {
	Configuration conf.KafkaConfiguration
	Producer      sarama.SyncProducer
}

Producer is an implementation of Producer interface

func New

func New(config *conf.ConfigStruct) (*Producer, error)

New constructs new implementation of Producer interface

func (*Producer) Close

func (producer *Producer) Close() error

Close allow the Sarama producer to be gracefully closed

func (*Producer) ProduceMessage

func (producer *Producer) ProduceMessage(msg types.ProducerMessage) (partitionID int32, offset int64, err error)

ProduceMessage produces message to selected topic. That function returns partition ID and offset of new message or an error value in case of any problem on broker side.

type SCRAMClient

type SCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

SCRAMClient implementation for the SCRAM authentication

func (*SCRAMClient) Begin

func (x *SCRAMClient) Begin(userName, password, authzID string) (err error)

Begin prepares the client for the SCRAM exchange

func (*SCRAMClient) Done

func (x *SCRAMClient) Done() bool

Done should return true when the SCRAM conversation is over.

func (*SCRAMClient) Step

func (x *SCRAMClient) Step(challenge string) (response string, err error)

Step steps client through the SCRAM exchange