kafka

Documentation

Index

• func NewConsumerGroup(brokers []string, groupID string, log logger.Logger) *consumerGroup
• func NewKafkaConn(ctx context.Context, kafkaCfg *Config) (*kafka.Conn, error)
• func NewKafkaReader(kafkaURL []string, topic, groupID string, errLogger kafka.Logger) *kafka.Reader
• func NewProducer(log logger.Logger, brokers []string) *producer
• func NewWriter(brokers []string, errLogger kafka.Logger) *kafka.Writer
• type Config
• type ConsumerGroup
• type MessageProcessor
• type Producer
• type TopicConfig
• type Worker

Functions

func NewConsumerGroup

func NewConsumerGroup(brokers []string, groupID string, log logger.Logger) *consumerGroup

NewConsumerGroup kafka consumer group constructor

func NewKafkaConn

func NewKafkaConn(ctx context.Context, kafkaCfg *Config) (*kafka.Conn, error)

NewKafkaConn create new kafka connection

func NewKafkaReader

func NewKafkaReader(kafkaURL []string, topic, groupID string, errLogger kafka.Logger) *kafka.Reader

NewKafkaReader create new configured kafka reader

func NewProducer

func NewProducer(log logger.Logger, brokers []string) *producer

NewProducer create new kafka producer

func NewWriter

func NewWriter(brokers []string, errLogger kafka.Logger) *kafka.Writer

NewWriter create new configured kafka writer

Types

type Config

type Config struct {
	Brokers    []string `mapstructure:"brokers"`
	GroupID    string   `mapstructure:"groupID"`
	InitTopics bool     `mapstructure:"initTopics"`
}

Config kafka config

type ConsumerGroup

type ConsumerGroup interface {
	ConsumeTopic(ctx context.Context, cancel context.CancelFunc, groupID, topic string, poolSize int, worker Worker)
	GetNewKafkaReader(kafkaURL []string, topic, groupID string) *kafka.Reader
	GetNewKafkaWriter(topic string) *kafka.Writer
}

type MessageProcessor

type MessageProcessor interface {
	ProcessMessages(ctx context.Context, r *kafka.Reader, wg *sync.WaitGroup, workerID int)
}

MessageProcessor processor methods must implement kafka.Worker func method interface

type Producer

type Producer interface {
	PublishMessage(ctx context.Context, msgs ...kafka.Message) error
	Close() error
}

type TopicConfig

type TopicConfig struct {
	TopicName         string `mapstructure:"topicName"`
	Partitions        int    `mapstructure:"partitions"`
	ReplicationFactor int    `mapstructure:"replicationFactor"`
}

TopicConfig kafka topic config

type Worker

type Worker func(ctx context.Context, r *kafka.Reader, wg *sync.WaitGroup, workerID int)

Worker kafka consumer worker fetch and process messages from reader