kafka

Documentation

Index

• func New(ctx context.Context, config Config) (*kafka.Conn, error)
• func NewKafkaReader(kafkaURL []string, topic, groupID string, errLogger kafka.Logger) *kafka.Reader
• func NewWriter(brokers []string, errLogger kafka.Logger) *kafka.Writer
• type Config
• type Consumer
  • func NewConsumer(brokers []string, groupID string, log logger.Logger) Consumer
• type Producer
  • func NewProducer(log logger.Logger, brokers []string) Producer
• type Worker

Functions

func New

func New(ctx context.Context, config Config) (*kafka.Conn, error)

func NewKafkaReader

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

NewKafkaReader create new configured kafka reader.

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
	GroupID string
}

type Consumer

type Consumer interface {
	ConsumeTopic(ctx context.Context, topics []string, poolSize int, worker Worker) error
}

func NewConsumer

func NewConsumer(brokers []string, groupID string, log logger.Logger) Consumer

NewConsumer kafka consumer constructor.

type Producer

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

func NewProducer

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

NewProducer create new queue producer.

type Worker

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

Worker kafka consumer worker fetch and process messages from reader.