consumer

Documentation

Index

• Constants
• func NewBatcher(input chan kafka.Message, size int, timeout time.Duration) *batcher
• func NewOffsetManager(logger log.Logger, reader Reader, batchSize int, batchTimeout time.Duration) *offsetManager
• func NewReader(logger log.Logger, dialer *kafka.Dialer, settings *Settings, ...) (*kafka.Reader, error)
• type Batcher
• type Consumer
  • func NewConsumer(ctx context.Context, conf cfg.Config, logger log.Logger, key string) (*Consumer, error)
  • func NewConsumerWithInterfaces(settings *Settings, logger log.Logger, manager OffsetManager) (*Consumer, error)
  • func (c *Consumer) Commit(ctx context.Context, msgs ...kafka.Message) error
  • func (c *Consumer) Data() chan kafka.Message
  • func (c *Consumer) Run(ctx context.Context) error
• type Offset
• type OffsetManager
• type Reader
• type ReaderOption
  • func WithBatch(maxSize int) ReaderOption
• type Settings
  • func ParseSettings(config cfg.Config, key string) *Settings
  • func (s *Settings) Connection() *connection.Settings
  • func (s *Settings) WithConnection(conn *connection.Settings) *Settings

Constants

const (
	// MaxBatchBytes is the Kafka default max batch size in bytes.
	MaxBatchBytes = 1000000

	// MaxBatchSize is the maximum batch size in number of messages.
	MaxBatchSize = 1024
)

const (
	// DefaultMaxRetryAttempts is how many times to retry a failed operation.
	DefaultMaxRetryAttempts = 3

	// DefaultConsumerGroupRetentionTime is the retention period of current offsets.
	DefaultConsumerGroupRetentionTime = 7 * 24 * time.Hour

	// DefaultMaxWait is a reasonable minimum for MaxWait.
	DefaultMaxWait = time.Second

	// CommitOffsetsSync == 0 means that auto-commit is disabled.
	CommitOffsetsSync = time.Duration(0)
)

Functions

func NewBatcher

func NewBatcher(input chan kafka.Message, size int, timeout time.Duration) *batcher

func NewOffsetManager

func NewOffsetManager(logger log.Logger, reader Reader, batchSize int, batchTimeout time.Duration) *offsetManager

func NewReader

func NewReader(
	logger log.Logger,
	dialer *kafka.Dialer,
	settings *Settings,
	opts ...ReaderOption,
) (*kafka.Reader, error)

Types

type Batcher

type Batcher interface {
	Get(ctx context.Context) []kafka.Message
}

type Consumer

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

func NewConsumer

func NewConsumer(
	ctx context.Context, conf cfg.Config, logger log.Logger, key string,
) (*Consumer, error)

func NewConsumerWithInterfaces

func NewConsumerWithInterfaces(settings *Settings, logger log.Logger, manager OffsetManager) (*Consumer, error)

func (*Consumer) Commit

func (c *Consumer) Commit(ctx context.Context, msgs ...kafka.Message) error

func (*Consumer) Data

func (c *Consumer) Data() chan kafka.Message

func (*Consumer) Run

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

type Offset

type Offset struct {
	Partition int
	Index     int64
}

type OffsetManager

type OffsetManager interface {
	Start(ctx context.Context) error
	Batch(ctx context.Context) []kafka.Message
	Commit(ctx context.Context, msgs ...kafka.Message) error
	Flush() error
}

type Reader

type Reader interface {
	FetchMessage(context.Context) (kafka.Message, error)
	ReadMessage(context.Context) (kafka.Message, error)
	CommitMessages(context.Context, ...kafka.Message) error
	Stats() kafka.ReaderStats
	Close() error
}

type ReaderOption

type ReaderOption func(*kafka.ReaderConfig)

func WithBatch

func WithBatch(maxSize int) ReaderOption

WithBatch sets batching configuration.

type Settings

type Settings struct {
	ConnectionName string `cfg:"connection" validate:"required"`

	Topic   string `cfg:"topic" validate:"required"`
	GroupID string `cfg:"group_id"`
	// FQTopic is the fully-qualified topic name (with prefix).
	FQTopic string
	// FQGroupID is the fully-qualified group id (with prefix).
	FQGroupID    string
	BatchSize    int           `cfg:"batch_size" default:"1"`
	BatchTimeout time.Duration `cfg:"idle_timeout" default:"1s"`
	// contains filtered or unexported fields
}

func ParseSettings

func ParseSettings(config cfg.Config, key string) *Settings

func (*Settings) Connection

func (s *Settings) Connection() *connection.Settings

func (*Settings) WithConnection

func (s *Settings) WithConnection(conn *connection.Settings) *Settings