easykafka

README

Usage

package main

import (
	easyKafka "github.com/diazoxide/easy-kafka"
	"log"
	"os"
)

type MyMessageType struct {
	From    string `json:"from"`
	Content string `json:"content"`
}

func main() {
	logger := log.New(os.Stdout, "test", log.Ltime)
	k := easyKafka.New[MyMessageType](
		[]string{"kafka:9092"}, // KafkaUris,
		"my-prefix", // Topics Prefix
		"my-group", // Consumer Group
		12, // Partitions
		logger, // Logger
	)
	k.Produce( "my-topic", &MyMessageType{"test","test"} )
	k.Consume([]string{"my-topic"}, func(message MyMessageType) error{
		
		// Your code here
		
		return nil
    }, true)
	// ...
}

Available methods

Consume

Produce

Documentation

Index

• type BaseProducer
  • func InitBaseProducer(brokers []string, opts ...BaseProducerOption) (producer *BaseProducer, close func() error)
  • func (p *BaseProducer) Produce(ctx context.Context, messages ...*kafka.Message) error
• type BaseProducerOption
  • func BaseProducerInitialPartitionsCount(count uint) BaseProducerOption
  • func BaseProducerWithErrorLogger(logger kafka.Logger) BaseProducerOption
  • func BaseProducerWithLogger(logger kafka.Logger) BaseProducerOption
  • func BaseProducerWithWriterConfig(writer *kafka.Writer) BaseProducerOption
• type Consumer
  • func InitConsumer[T any](brokers []string, topicsList []string, groupId string, ...) (consumer *Consumer[T], close func() error)
  • func (k *Consumer[T]) Consume(ctx context.Context, handler ConsumerHandler[T])
• type ConsumerErrorHandler
• type ConsumerHandler
• type ConsumerOption
  • func ConsumerConcurrency[T any](concurrency uint) ConsumerOption[T]
  • func ConsumerDynamicTopicsDiscovery[T any]() ConsumerOption[T]
  • func ConsumerDynamicTopicsDiscoveryInterval[T any](interval time.Duration) ConsumerOption[T]
  • func ConsumerInitialPartitionsCount[T any](count uint) ConsumerOption[T]
  • func ConsumerTopicNamesExactMatch[T any]() ConsumerOption[T]
  • func ConsumerTopicNamesRegexMatch[T any]() ConsumerOption[T]
  • func ConsumerWithErrorLogger[T any](logger kafka.Logger) ConsumerOption[T]
  • func ConsumerWithLogger[T any](logger kafka.Logger) ConsumerOption[T]
  • func ConsumerWithMaxBlockingTasks[T any](count uint) ConsumerOption[T]
  • func ConsumerWithOnFailCommitHandler[T any](handler ConsumerErrorHandler[T]) ConsumerOption[T]
  • func ConsumerWithReadMessageHandler[T any](handler ConsumerErrorHandler[T]) ConsumerOption[T]
  • func ConsumerWithReaderConfig[T any](config kafka.ReaderConfig) ConsumerOption[T]
  • func ConsumerWithTopicsListUpdatedHandler[T any](handler ConsumerTopicsListUpdatedHandler[T]) ConsumerOption[T]
  • func ConsumerWithWrongMessageHandler[T any](handler ConsumerErrorHandler[T]) ConsumerOption[T]
• type ConsumerTopicsListUpdatedHandler
• type ErrorHandler
• type LoggerContainer
• type Producer
  • func InitProducer[T any](brokers []string, opts ...BaseProducerOption) (producer *Producer[T], close func() error)
  • func (p *Producer[T]) Produce(ctx context.Context, topics []string, messages ...*T) error
• type Stream
  • func InitStream[T any](brokers []string, topicsList []string, groupId string, opts ...StreamOption[T]) (*Stream[T], func() error)
  • func (s *Stream[T]) Run(ctx context.Context, routingRules ...StreamRoutingRule[T])
• type StreamOption
  • func StreamWithConsumerOptions[T any](opts ...ConsumerOption[T]) StreamOption[T]
  • func StreamWithErrorLogger[T any](logger kafka.Logger) StreamOption[T]
  • func StreamWithLogger[T any](logger kafka.Logger) StreamOption[T]
  • func StreamWithParallelJobs[T any](parallelJobs uint) StreamOption[T]
  • func StreamWithProducerOptions[T any](opts ...BaseProducerOption) StreamOption[T]
• type StreamRoutingRule

Types

type BaseProducer

type BaseProducer struct {
	LoggerContainer
	// contains filtered or unexported fields
}

BaseProducer is a wrapper around kafka.Writer

func InitBaseProducer

func InitBaseProducer(
	brokers []string,
	opts ...BaseProducerOption,
) (producer *BaseProducer, close func() error)

InitBaseProducer creates a new producer instance

func (*BaseProducer) Produce

func (p *BaseProducer) Produce(ctx context.Context, messages ...*kafka.Message) error

Produce writes messages to kafka

type BaseProducerOption

type BaseProducerOption func(kafka *BaseProducer) error

BaseProducerOption is a function that sets some option on the producer

func BaseProducerInitialPartitionsCount

func BaseProducerInitialPartitionsCount(count uint) BaseProducerOption

BaseProducerInitialPartitionsCount sets initial partitions count

func BaseProducerWithErrorLogger

func BaseProducerWithErrorLogger(logger kafka.Logger) BaseProducerOption

BaseProducerWithErrorLogger sets error logger

func BaseProducerWithLogger

func BaseProducerWithLogger(logger kafka.Logger) BaseProducerOption

BaseProducerWithLogger sets logger

func BaseProducerWithWriterConfig

func BaseProducerWithWriterConfig(writer *kafka.Writer) BaseProducerOption

BaseProducerWithWriterConfig sets writer config

type Consumer

type Consumer[T any] struct {
	LoggerContainer
	// contains filtered or unexported fields
}

Consumer is a wrapper around kafka.Reader

func InitConsumer

func InitConsumer[T any](
	brokers []string,
	topicsList []string,
	groupId string,
	opts ...ConsumerOption[T],
) (consumer *Consumer[T], close func() error)

InitConsumer creates a new consumer instance

func (*Consumer[T]) Consume

func (k *Consumer[T]) Consume(ctx context.Context, handler ConsumerHandler[T])

Consume starts consuming messages from kafka

type ConsumerErrorHandler

type ConsumerErrorHandler[T any] func(k *Consumer[T], err error)

type ConsumerHandler

type ConsumerHandler[T any] func(message *T, kafkaMessage *kafka.Message) error

ConsumerHandler is a function that handles messages

type ConsumerOption

type ConsumerOption[T any] func(kafka *Consumer[T]) error

ConsumerOption is a function that sets some option

func ConsumerConcurrency

func ConsumerConcurrency[T any](concurrency uint) ConsumerOption[T]

ConsumerConcurrency sets parallel tasks count

func ConsumerDynamicTopicsDiscovery

func ConsumerDynamicTopicsDiscovery[T any]() ConsumerOption[T]

ConsumerDynamicTopicsDiscovery enable dynamic topics discovery

func ConsumerDynamicTopicsDiscoveryInterval

func ConsumerDynamicTopicsDiscoveryInterval[T any](interval time.Duration) ConsumerOption[T]

ConsumerDynamicTopicsDiscoveryInterval sets dynamic topics discovery interval

func ConsumerInitialPartitionsCount

func ConsumerInitialPartitionsCount[T any](count uint) ConsumerOption[T]

ConsumerInitialPartitionsCount sets initial partitions count

func ConsumerTopicNamesExactMatch

func ConsumerTopicNamesExactMatch[T any]() ConsumerOption[T]

ConsumerTopicNamesExactMatch enable exact match for topic names

func ConsumerTopicNamesRegexMatch

func ConsumerTopicNamesRegexMatch[T any]() ConsumerOption[T]

ConsumerTopicNamesRegexMatch enable regex match for topic names

func ConsumerWithErrorLogger

func ConsumerWithErrorLogger[T any](logger kafka.Logger) ConsumerOption[T]

ConsumerWithErrorLogger sets error logger

func ConsumerWithLogger

func ConsumerWithLogger[T any](logger kafka.Logger) ConsumerOption[T]

ConsumerWithLogger sets logger

func ConsumerWithMaxBlockingTasks

func ConsumerWithMaxBlockingTasks[T any](count uint) ConsumerOption[T]

ConsumerWithMaxBlockingTasks sets max blocking tasks

func ConsumerWithOnFailCommitHandler

func ConsumerWithOnFailCommitHandler[T any](handler ConsumerErrorHandler[T]) ConsumerOption[T]

ConsumerWithOnFailCommitHandler sets handler for commit error

func ConsumerWithReadMessageHandler

func ConsumerWithReadMessageHandler[T any](handler ConsumerErrorHandler[T]) ConsumerOption[T]

ConsumerWithReadMessageHandler sets handler for read message

func ConsumerWithReaderConfig

func ConsumerWithReaderConfig[T any](config kafka.ReaderConfig) ConsumerOption[T]

ConsumerWithReaderConfig sets reader config

func ConsumerWithTopicsListUpdatedHandler

func ConsumerWithTopicsListUpdatedHandler[T any](handler ConsumerTopicsListUpdatedHandler[T]) ConsumerOption[T]

ConsumerWithTopicsListUpdatedHandler sets handler for topics list update

func ConsumerWithWrongMessageHandler

func ConsumerWithWrongMessageHandler[T any](handler ConsumerErrorHandler[T]) ConsumerOption[T]

ConsumerWithWrongMessageHandler sets handler for wrong message

type ConsumerTopicsListUpdatedHandler

type ConsumerTopicsListUpdatedHandler[T any] func(k *Consumer[T], topics []string)

type ErrorHandler

type ErrorHandler[T any] func(k *Consumer[T], err error)

ErrorHandler is a function that handles errors

type LoggerContainer

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

type Producer

type Producer[T any] struct {
	*BaseProducer
}

Producer is a wrapper around BaseProducer

func InitProducer

func InitProducer[T any](
	brokers []string,
	opts ...BaseProducerOption,
) (producer *Producer[T], close func() error)

InitProducer initializes a new Producer instance

func (*Producer[T]) Produce

func (p *Producer[T]) Produce(ctx context.Context, topics []string, messages ...*T) error

Produce sends messages to kafka

type Stream

type Stream[T any] struct {
	LoggerContainer
	// contains filtered or unexported fields
}

Stream is a wrapper around Consumer and Producer

func InitStream

func InitStream[T any](
	brokers []string,
	topicsList []string,
	groupId string,
	opts ...StreamOption[T],
) (*Stream[T], func() error)

InitStream initializes a new Stream instance

func (*Stream[T]) Run

func (s *Stream[T]) Run(ctx context.Context, routingRules ...StreamRoutingRule[T])

Run starts the stream

type StreamOption

type StreamOption[T any] func(stream *Stream[T]) error

StreamOption is a function that modifies a Stream instance

func StreamWithConsumerOptions

func StreamWithConsumerOptions[T any](opts ...ConsumerOption[T]) StreamOption[T]

StreamWithConsumerOptions sets consumer options

func StreamWithErrorLogger

func StreamWithErrorLogger[T any](logger kafka.Logger) StreamOption[T]

StreamWithErrorLogger sets error logger

func StreamWithLogger

func StreamWithLogger[T any](logger kafka.Logger) StreamOption[T]

StreamWithLogger sets logger

func StreamWithParallelJobs

func StreamWithParallelJobs[T any](parallelJobs uint) StreamOption[T]

StreamWithParallelJobs sets parallel jobs count for stream consumer

func StreamWithProducerOptions

func StreamWithProducerOptions[T any](opts ...BaseProducerOption) StreamOption[T]

StreamWithProducerOptions sets producer options

type StreamRoutingRule

type StreamRoutingRule[T any] func(message *T, kafkaMessage *kafka.Message) (newMessage *kafka.Message, err error)

StreamRoutingRule is a function that modifies a message