kafka

package

v0.0.0-...-0c44e78 Latest Latest Go to latest Published: Mar 19, 2024 License: MIT Imports: 6 Imported by: 0

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/scul0405/saga-orchestration

Links

Open Source Insights

Documentation ¶

Index ¶

func NewKafkaConn(addr string) (*kafka.Conn, error)
func NewKafkaReader(kafkaURL []string, topic, groupID string) *kafka.Reader
func NewKafkaWriter(brokers []string) *kafka.Writer
type ConsumerGroup
- func NewConsumerGroup(brokers []string, log logger.Logger) ConsumerGroup
type Producer
- func NewProducer(log logger.Logger, brokers []string) Producer
type Worker

Constants ¶

This section is empty.

Variables ¶

This section is empty.

Functions ¶

func NewKafkaConn ¶

func NewKafkaConn(addr string) (*kafka.Conn, error)

func NewKafkaReader ¶

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

func NewKafkaWriter ¶

func NewKafkaWriter(brokers []string) *kafka.Writer

Types ¶

type ConsumerGroup ¶

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

func NewConsumerGroup ¶

func NewConsumerGroup(brokers []string, log logger.Logger) ConsumerGroup

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 kafka producer

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

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL