kafka

Documentation

Index

• func BuildConsumerGroupConfig(conf *Config, initial int64) (*sarama.Config, error)
• func BuildProducerConfig(conf Config) (*sarama.Config, error)
• func Check(ctx context.Context, conf *Config, topics []string) error
• func GetContextWithMQHeader(header []*sarama.RecordHeader) context.Context
• func GetMQHeaderWithContext(ctx context.Context) ([]sarama.RecordHeader, error)
• func NewConsumerGroup(conf *sarama.Config, addr []string, groupID string) (sarama.ConsumerGroup, error)
• func NewProducer(conf *sarama.Config, addr []string) (sarama.SyncProducer, error)
• type Config
• type MConsumerGroup
  • func NewMConsumerGroup(conf *Config, groupID string, topics []string) (*MConsumerGroup, error)
  • func (mc *MConsumerGroup) Close() error
  • func (mc *MConsumerGroup) GetContextFromMsg(cMsg *sarama.ConsumerMessage) context.Context
  • func (mc *MConsumerGroup) RegisterHandleAndConsumer(ctx context.Context, handler sarama.ConsumerGroupHandler)
• type Producer
  • func NewKafkaProducer(config *sarama.Config, addr []string, topic string) (*Producer, error)
  • func (p *Producer) SendMessage(ctx context.Context, key string, msg proto.Message) (int32, int64, error)
• type TLSConfig

Functions

func BuildConsumerGroupConfig

func BuildConsumerGroupConfig(conf *Config, initial int64) (*sarama.Config, error)

func BuildProducerConfig

func BuildProducerConfig(conf Config) (*sarama.Config, error)

func Check

func Check(ctx context.Context, conf *Config, topics []string) error

func GetContextWithMQHeader

func GetContextWithMQHeader(header []*sarama.RecordHeader) context.Context

GetContextWithMQHeader creates a context from message queue headers.

func GetMQHeaderWithContext

func GetMQHeaderWithContext(ctx context.Context) ([]sarama.RecordHeader, error)

GetMQHeaderWithContext extracts message queue headers from the context.

func NewConsumerGroup

func NewConsumerGroup(conf *sarama.Config, addr []string, groupID string) (sarama.ConsumerGroup, error)

func NewProducer

func NewProducer(conf *sarama.Config, addr []string) (sarama.SyncProducer, error)

Types

type Config

type Config struct {
	Username     string    `yaml:"username"`
	Password     string    `yaml:"password"`
	ProducerAck  string    `yaml:"producerAck"`
	CompressType string    `yaml:"compressType"`
	Addr         []string  `yaml:"addr"`
	TLS          TLSConfig `yaml:"tls"`
}

type MConsumerGroup

type MConsumerGroup struct {
	sarama.ConsumerGroup
	// contains filtered or unexported fields
}

func NewMConsumerGroup

func NewMConsumerGroup(conf *Config, groupID string, topics []string) (*MConsumerGroup, error)

func (*MConsumerGroup) Close

func (mc *MConsumerGroup) Close() error

func (*MConsumerGroup) GetContextFromMsg

func (mc *MConsumerGroup) GetContextFromMsg(cMsg *sarama.ConsumerMessage) context.Context

func (*MConsumerGroup) RegisterHandleAndConsumer

func (mc *MConsumerGroup) RegisterHandleAndConsumer(ctx context.Context, handler sarama.ConsumerGroupHandler)

type Producer

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

Producer represents a Kafka producer.

func NewKafkaProducer

func NewKafkaProducer(config *sarama.Config, addr []string, topic string) (*Producer, error)

func (*Producer) SendMessage

func (p *Producer) SendMessage(ctx context.Context, key string, msg proto.Message) (int32, int64, error)

SendMessage sends a message to the Kafka topic configured in the Producer.

type TLSConfig

type TLSConfig struct {
	EnableTLS          bool   `yaml:"enableTLS"`
	CACrt              string `yaml:"caCrt"`
	ClientCrt          string `yaml:"clientCrt"`
	ClientKey          string `yaml:"clientKey"`
	ClientKeyPwd       string `yaml:"clientKeyPwd"`
	InsecureSkipVerify bool   `yaml:"insecureSkipVerify"`
}