consumer

Documentation

Index

• type ConsumeProcessor
• type Listener
• type ListenerOptionFn
  • func WithBatchSize(batchSize int) ListenerOptionFn
  • func WithConsumeTimeout(consumeTimeout time.Duration) ListenerOptionFn
  • func WithRepeatStrategyEven(maxAttempts int, intervalSec int) ListenerOptionFn
  • func WithRepeatStrategyProgressive(maxAttempts int, intervalSec int, multiplier float32) ListenerOptionFn
• type ProcessResult
• type RepeatStrategy
• type RepeatStrategyEven
• type RepeatStrategyProgressive
• type Service
  • func New(host string, port int, options ...ServiceOptionFn) *Service
  • func (c *Service) Consume(ctx context.Context, topic, group string, processor ConsumeProcessor, ...) error
• type ServiceOptionFn
  • func WithServiceUnavailableTimeout(unavailableTimeout time.Duration) ServiceOptionFn

Types

type ConsumeProcessor

type ConsumeProcessor = func(ctx context.Context, data []byte, id string) error

type Listener

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

type ListenerOptionFn

type ListenerOptionFn = func(c *Listener)

func WithBatchSize

func WithBatchSize(batchSize int) ListenerOptionFn

func WithConsumeTimeout

func WithConsumeTimeout(consumeTimeout time.Duration) ListenerOptionFn

func WithRepeatStrategyEven

func WithRepeatStrategyEven(maxAttempts int, intervalSec int) ListenerOptionFn

func WithRepeatStrategyProgressive

func WithRepeatStrategyProgressive(maxAttempts int, intervalSec int, multiplier float32) ListenerOptionFn

type ProcessResult

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

type RepeatStrategy

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

type RepeatStrategyEven

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

type RepeatStrategyProgressive

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

type Service

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

func New

func New(host string, port int, options ...ServiceOptionFn) *Service

func (*Service) Consume

func (c *Service) Consume(ctx context.Context, topic, group string, processor ConsumeProcessor, options ...ListenerOptionFn) error

type ServiceOptionFn

type ServiceOptionFn = func(c *Service)

func WithServiceUnavailableTimeout

func WithServiceUnavailableTimeout(unavailableTimeout time.Duration) ServiceOptionFn