kafkax

Documentation

Index

• Variables
• func NewConsumer(config *Config) (*kafka.Consumer, error)
• func NewConsumers(configs Configs) (map[string]*kafka.Consumer, error)
• func NewProducer(config *Config) (*kafka.Producer, error)
• func NewProducers(configs Configs) (map[string]*kafka.Producer, error)
• type Config
• type Configs

Variables

var Provider = wire.NewSet(
	NewConsumers,
	NewProducers,
)

Functions

func NewConsumer

func NewConsumer(config *Config) (*kafka.Consumer, error)

func NewConsumers

func NewConsumers(configs Configs) (map[string]*kafka.Consumer, error)

func NewProducer

func NewProducer(config *Config) (*kafka.Producer, error)

func NewProducers

func NewProducers(configs Configs) (map[string]*kafka.Producer, error)

Types

type Config

type Config struct {
	Topic                     string   `mapstructure:"topic" json:"topic" yaml:"topic"`
	Brokers                   []string `mapstructure:"brokers" json:"brokers" yaml:"brokers"`
	GroupID                   string   `mapstructure:"group_id" json:"group_id" yaml:"group_id"`
	SecurityProtocol          string   `mapstructure:"security_protocol" json:"security_protocol" yaml:"security_protocol"`
	SaslMechanism             string   `mapstructure:"sasl_mechanism" json:"sasl_mechanism" yaml:"sasl_mechanism"`
	SaslUsername              string   `mapstructure:"sasl_username" json:"sasl_username" yaml:"sasl_username"`
	SaslPassword              string   `mapstructure:"sasl_password" json:"sasl_password" yaml:"sasl_password"`
	SslCaLocation             string   `mapstructure:"ssl_ca_location" json:"ssl_ca_location" yaml:"ssl_ca_location"`
	EnableSslCertVerification bool     `mapstructure:"enable_ssl_cert_verification" json:"enable_ssl_cert_verification" yaml:"enable_ssl_cert_verification"`
}

type Configs

type Configs map[string]*Config