reader

Documentation

Index

• func DefaultConfig() kafka.ReaderConfig
• func NewReader(opts ...Opt) *kafka.Reader
• type Opt
  • func WithBrokers(brokers ...string) Opt
  • func WithDialer(dialer *kafka.Dialer) Opt
  • func WithErrorLogger(logger kafka.Logger) Opt
  • func WithGroupID(id string) Opt
  • func WithLogger(logger kafka.Logger) Opt
  • func WithTopic(topic string) Opt
• type Reader

Functions

func DefaultConfig

func DefaultConfig() kafka.ReaderConfig

DefaultConfig returns the default configuration for a Kafka reader.

func NewReader

func NewReader(opts ...Opt) *kafka.Reader

NewReader creates a new Kafka reader.

Types

type Opt

type Opt func(*kafka.ReaderConfig)

Opt defines an option for a Kafka reader.

func WithBrokers

func WithBrokers(brokers ...string) Opt

WithBrokers configures the brokers for the reader.

func WithDialer

func WithDialer(dialer *kafka.Dialer) Opt

WithDialer defines a dialer for the reader.

func WithErrorLogger

func WithErrorLogger(logger kafka.Logger) Opt

WithErrorLogger configures the error logger for the reader.

func WithGroupID

func WithGroupID(id string) Opt

WithGroupID defines the consumer group ID.

func WithLogger

func WithLogger(logger kafka.Logger) Opt

WithLogger configures the logger for the reader.

func WithTopic

func WithTopic(topic string) Opt

WithTopic defines the topic for the reader.

type Reader

type Reader interface {
	// Close closes the reader.
	Close() error

	// CommitMessages commits the messages to Kafka.
	CommitMessages(ctx context.Context, msgs ...kafka.Message) error

	// Config returns the reader configuration.
	Config() kafka.ReaderConfig

	// FetchMessage fetches a message from Kafka.
	FetchMessage(ctx context.Context) (kafka.Message, error)

	// Lag returns the current lag of the reader.
	Lag() int64

	// Offset returns the current offset of the reader.
	Offset() int64

	// ReadLag reads the current lag of the reader.
	ReadLag(ctx context.Context) (lag int64, err error)

	// ReadMessage reads a message from Kafka.
	ReadMessage(ctx context.Context) (kafka.Message, error)

	// SetOffset sets the offset of the reader.
	SetOffset(offset int64) error

	// SetOffsetAt sets the offset of the reader at the given time.
	SetOffsetAt(ctx context.Context, t time.Time) error

	// Stats returns the reader statistics.
	Stats() kafka.ReaderStats
}

Reader ...