kafka

README

go-watermillx-driver-kafka

🚀 Install

go get github.com/cyg-pd/go-watermillx-driver-kafka@v1

This library is v1 and follows SemVer strictly.

No breaking changes will be made to exported APIs before v2.0.0.

💡 Usage

You can import go-watermillx-driver-kafka using:

package main

import (
	_ "github.com/cyg-pd/go-watermillx"
	"github.com/cyg-pd/go-watermillx-driver-kafka"
)

func main() {
	drv, err := driver.New("kafka", `{"Brokers": ["localhost:9092"], "InitializeTopicDetails":{"NumPartitions": 3}}`)
	if err != nil {
		panic(err)
	}

	pub, err := drv.Publisher()
	if err != nil {
		panic(err)
	}

	sub, err := drv.Subscriber()
	if err != nil {
		panic(err)
	}
}

Documentation

Index

• func DefaultSaramaConfig() *sarama.Config
• type Kafka
  • func New(config any) (*Kafka, error)
  • func (k *Kafka) Publisher() (message.Publisher, error)
  • func (k *Kafka) Subscriber(opts ...driver.SubscriberOption) (message.Subscriber, error)
• type KafkaConfig
  • func NewKafkaConfig() *KafkaConfig

Functions

func DefaultSaramaConfig

func DefaultSaramaConfig() *sarama.Config

Types

type Kafka

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

func New

func New(config any) (*Kafka, error)

func (*Kafka) Publisher

func (k *Kafka) Publisher() (message.Publisher, error)

func (*Kafka) Subscriber

func (k *Kafka) Subscriber(opts ...driver.SubscriberOption) (message.Subscriber, error)

type KafkaConfig

type KafkaConfig struct {
	// Kafka brokers list.
	Brokers []string

	// SaramaConfig holds additional sarama settings.
	SaramaConfig *sarama.Config

	Subscriber struct {
		// Kafka consumer group.
		// When empty, all messages from all partitions will be returned.
		ConsumerGroup string
	}

	InitializeTopicDetails *sarama.TopicDetail
}

func NewKafkaConfig

func NewKafkaConfig() *KafkaConfig