kafka

Documentation

Overview

Package kafka provides a simple high-level API for producing and consuming Apache Kafka messages.

Based on github.com/segmentio/kafka-go, it abstracts away the complexities of the Kafka protocol and provides a simplified interface for working with Kafka topics.

It allows to specify custom message encoding and decoding functions, including serialization and encryption.

Index

• func DefaultMessageDecodeFunc(_ context.Context, msg []byte, data any) error
• func DefaultMessageEncodeFunc(_ context.Context, data any) ([]byte, error)
• type Consumer
  • func NewConsumer(brokers []string, topic, groupID string, opts ...Option) (*Consumer, error)
  • func (c *Consumer) Close() error
  • func (c *Consumer) HealthCheck(ctx context.Context) error
  • func (c *Consumer) Receive(ctx context.Context) ([]byte, error)
  • func (c *Consumer) ReceiveData(ctx context.Context, data any) error
• type Option
  • func WithFirstOffset() Option
  • func WithMessageDecodeFunc(f TDecodeFunc) Option
  • func WithMessageEncodeFunc(f TEncodeFunc) Option
  • func WithSessionTimeout(t time.Duration) Option
• type Producer
  • func NewProducer(urls []string, topic string, opts ...Option) (*Producer, error)
  • func (p *Producer) Close() error
  • func (p *Producer) Send(ctx context.Context, msg []byte) error
  • func (p *Producer) SendData(ctx context.Context, data any) error
• type TDecodeFunc
• type TEncodeFunc

Functions

func DefaultMessageDecodeFunc

func DefaultMessageDecodeFunc(_ context.Context, msg []byte, data any) error

DefaultMessageDecodeFunc is the default function to decode a message for ReceiveData(). The value underlying data must be a pointer to the correct type for the next data item received.

func DefaultMessageEncodeFunc

func DefaultMessageEncodeFunc(_ context.Context, data any) ([]byte, error)

DefaultMessageEncodeFunc is the default function to encode the input data for SendData().

Types

type Consumer

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

Consumer represents a wrapper around kafka.Consumer.

func NewConsumer

func NewConsumer(brokers []string, topic, groupID string, opts ...Option) (*Consumer, error)

NewConsumer creates a new instance of Consumer. Please call the HealthCheck() method to check if the connection is working.

func (*Consumer) Close

func (c *Consumer) Close() error

Close cleans up Consumer's internal resources.

func (*Consumer) HealthCheck

func (c *Consumer) HealthCheck(ctx context.Context) error

HealthCheck checks if the consumer is working.

func (*Consumer) Receive

func (c *Consumer) Receive(ctx context.Context) ([]byte, error)

Receive reads one message from the Kafka; blocks if there are no messages in the queue.

func (*Consumer) ReceiveData

func (c *Consumer) ReceiveData(ctx context.Context, data any) error

ReceiveData retrieves a message from the queue and extract its content in the data.

type Option

type Option func(*config)

Option is a type alias for a function that configures Kafka client.

func WithFirstOffset

func WithFirstOffset() Option

WithFirstOffset tells Kafka consumer to read from the beginning of uncommitted offset. By default it will read from the end of the queue.

func WithMessageDecodeFunc

func WithMessageDecodeFunc(f TDecodeFunc) Option

WithMessageDecodeFunc allow to replace DefaultMessageDecodeFunc(). This function used by ReceiveData() to decode a message encoded with messageEncodeFunc to the provided data object. The value underlying data must be a pointer to the correct type for the next data item received.

func WithMessageEncodeFunc

func WithMessageEncodeFunc(f TEncodeFunc) Option

WithMessageEncodeFunc allow to replace DefaultMessageEncodeFunc. This function used by SendData() to encode the input data.

func WithSessionTimeout

func WithSessionTimeout(t time.Duration) Option

WithSessionTimeout sets the timeout used to detect client failures when using Kafka's group management facility. The client sends periodic heartbeats to indicate its liveness to the broker. If no heartbeats are received by the broker before the expiration of this session timeout, then the broker will remove this client from the group and initiate a rebalance.

type Producer

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

Producer represents a wrapper around kafka.Producer.

func NewProducer

func NewProducer(urls []string, topic string, opts ...Option) (*Producer, error)

NewProducer creates a new instance of Producer.

func (*Producer) Close

func (p *Producer) Close() error

Close cleans up Producer's internal resources.

func (*Producer) Send

func (p *Producer) Send(ctx context.Context, msg []byte) error

Send sends a message to Kafka topic.

func (*Producer) SendData

func (p *Producer) SendData(ctx context.Context, data any) error

SendData delivers the specified data as encoded message to the queue.

type TDecodeFunc

type TDecodeFunc func(ctx context.Context, msg []byte, data any) error

TDecodeFunc is the type of function used to replace the default message decoding function used by ReceiveData().

type TEncodeFunc

type TEncodeFunc func(ctx context.Context, data any) ([]byte, error)

TEncodeFunc is the type of function used to replace the default message encoding function used by SendData().