producer

Documentation

Index

• type Config
• type Producer
  • func New(ctx context.Context, wg *sync.WaitGroup, topic string, config Config) (*Producer, error)
  • func (p *Producer) HealthCheck() error
  • func (p *Producer) Produce(key, value []byte) error
  • func (p *Producer) ProduceSync(key, value []byte) error
• type SSLConfig

Types

type Config

type Config struct {
	Idempotence      bool
	ReadEvents       bool
	FlushTimeoutMs   int
	BootstrapServers string
	SSL              SSLConfig
}

type Producer

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

func New

func New(ctx context.Context, wg *sync.WaitGroup, topic string, config Config) (*Producer, error)

func (*Producer) HealthCheck

func (p *Producer) HealthCheck() error

func (*Producer) Produce

func (p *Producer) Produce(key, value []byte) error

It is an example of how we can implement the message send to Kafka topic. Feel free to change this implementation if you need to add more options like headers, timestamp, partition etc.

func (*Producer) ProduceSync

func (p *Producer) ProduceSync(key, value []byte) error

The same as Produce but imitates synchronous send

type SSLConfig

type SSLConfig struct {
	Enabled             bool
	KeyLocation         string
	CertificateLocation string
	CALocation          string
}