confluent

Documentation

Index

• func New[T any](consumer *kafka.Consumer) function.CmdFunc[T]
• type Helper
  • func NewHelper[T any](consumer *kafka.Consumer, handler function.Handler[T]) *Helper[T]
  • func NewHelperWithOptions[T any](consumer *kafka.Consumer, handler function.Handler[T], options *Options) *Helper[T]
  • func (h *Helper[T]) Start()
• type Options
  • func DefaultOptions() (*Options, error)
• type Subscriber
  • func NewSubscriber[T any](consumer *kafka.Consumer, handler function.Handler[T], options *Options) *Subscriber[T]
  • func (l *Subscriber[T]) Subscribe(ctx context.Context) error

Functions

func New

func New[T any](consumer *kafka.Consumer) function.CmdFunc[T]

New returns CmdFunc for cloudevents command.

Types

type Helper

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

Helper assists in creating event handlers.

func NewHelper

func NewHelper[T any](consumer *kafka.Consumer, handler function.Handler[T]) *Helper[T]

NewHelper returns a new Helper with default options.

func NewHelperWithOptions

func NewHelperWithOptions[T any](consumer *kafka.Consumer, handler function.Handler[T], options *Options) *Helper[T]

NewHelperWithOptions returns a new Helper with options.

func (*Helper[T]) Start

func (h *Helper[T]) Start()

type Options

type Options struct {
	Topics       []string
	TimeOut      time.Duration
	ManualCommit bool
	MaxWorkers   int64
	Backoff      bool
	BackoffBase  time.Duration // Base duration for backoff
	MaxBackoff   time.Duration // Maximum backoff duration
	RetryLimit   int           // Limit for retries (-1 for infinite retries)
}

Options can be used to create customized handler.

func DefaultOptions

func DefaultOptions() (*Options, error)

DefaultOptions returns options based in config.

type Subscriber

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

Subscriber represents a subscriber listener.

func NewSubscriber

func NewSubscriber[T any](consumer *kafka.Consumer, handler function.Handler[T], options *Options) *Subscriber[T]

NewSubscriber returns a subscriber listener.

func (*Subscriber[T]) Subscribe

func (l *Subscriber[T]) Subscribe(ctx context.Context) error

Subscribe subscribes to specific topics and processes messages by partition.