statelessretry

Documentation

Index

• func ConsumeWithRetry(ctx context.Context, option *ConsumerWithRetryOption)
• type ConsumerWithRetryOption
• type ErrKafkaRetrieble
  • func (e *ErrKafkaRetrieble) Error() string
• type Handler

Functions

func ConsumeWithRetry

func ConsumeWithRetry(ctx context.Context, option *ConsumerWithRetryOption)

Types

type ConsumerWithRetryOption

type ConsumerWithRetryOption struct {
	Handler       Handler
	Consumer      *kafka.Consumer
	RetryQueue    chan *kafka.Message // make sure you passed buffered channel to get concurrent retries otherwies new messages will wait unitl previous retries are finished
	MaxRetryCount int
	Backoff       backoff.BackOff
}

type ErrKafkaRetrieble

type ErrKafkaRetrieble struct {
	Err error
}

func (*ErrKafkaRetrieble) Error

func (e *ErrKafkaRetrieble) Error() string

type Handler

type Handler interface {
	Hanle(msg *kafka.Message) error
	MoveToDQL(msg *kafka.Message) error
}