kafka

README

kafka-do

v0.3.4

kafka-do

What

Higher level abstraction for franz-go.

Why

We want to be able to write our kafka applications without making the same things over and over.

Batch Consume
Consume messages as much as you defined.

Batch Produce
Produce messages as a batch to a topic.

Example

For e2e example, check here.

	producer, err := kafka.NewProducer("127.0.0.1:9092")
	if err != nil {
		log.Fatal(err)
	}
	defer producer.Close()

	producer.Produce(context.Background(), []kafka.Message{
		kafka.Message("message 1"),
		kafka.Message("message 2"),
		kafka.Message("message 3"),
		kafka.Message("message 4"),
	}, "messages")

	consumer, err := kafka.NewConsumer("kafka_do", []string{"messages"}, []string{"127.0.0.1:9092"})
	if err != nil {
		log.Fatal(err)
	}
	defer consumer.Close()

	messages, errs := consumer.ConsumeBatch(context.Background(), 2)
	for _, message := range messages {
		log.Println(message)
	}

	for _, err := range errs {
		log.Println(err)
	}

Development

To run tests, start a kafka that runs on ":9092".

go test ./... -v -cover -count=1 -race

Documentation

Index

• type Consumer
  • func NewConsumer(groupName string, topics []string, brokers []string, logger bool) (*Consumer, error)
  • func (c *Consumer) Close()
  • func (c *Consumer) ConsumeBatch(ctx context.Context, batchSize int) []Message
• type Message
• type Producer
  • func NewProducer(maxBytes int32, brokers ...string) (*Producer, error)
  • func (p *Producer) Close()
  • func (p *Producer) Produce(ctx context.Context, messages []Message, topic string) kgo.ProduceResults

Types

type Consumer

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

func NewConsumer

func NewConsumer(groupName string, topics []string, brokers []string, logger bool) (*Consumer, error)

func (*Consumer) Close

func (c *Consumer) Close()

func (*Consumer) ConsumeBatch

func (c *Consumer) ConsumeBatch(ctx context.Context, batchSize int) []Message

type Message

type Message []byte

type Producer

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

func NewProducer

func NewProducer(maxBytes int32, brokers ...string) (*Producer, error)

func (*Producer) Close

func (p *Producer) Close()

func (*Producer) Produce

func (p *Producer) Produce(ctx context.Context, messages []Message, topic string) kgo.ProduceResults