handler

Documentation

Index

• type ConsumerMetricStorage
• type Middleware
  • func Log(logger log.Logger) Middleware
  • func Metrics(metricStorage ConsumerMetricStorage) Middleware
• type Result
  • func Commit() Result
  • func Retry(after time.Duration, err error) Result
• type Sync
  • func NewSync(logger log.Logger, adapter SyncHandlerAdapter, middlewares ...Middleware) Sync
  • func (r Sync) Handle(ctx context.Context, delivery *consumer.Delivery)
• type SyncHandlerAdapter
• type SyncHandlerAdapterFunc
  • func (a SyncHandlerAdapterFunc) Handle(ctx context.Context, delivery *consumer.Delivery) Result

Types

type ConsumerMetricStorage

type ConsumerMetricStorage interface {
	ObserveConsumeDuration(consumerGroup, topic string, t time.Duration)
	ObserveConsumeMsgSize(consumerGroup, topic string, size int)
	IncCommitCount(consumerGroup, topic string)
	IncRetryCount(consumerGroup, topic string)
}

type Middleware

type Middleware func(next SyncHandlerAdapter) SyncHandlerAdapter

func Log

func Log(logger log.Logger) Middleware

func Metrics

func Metrics(metricStorage ConsumerMetricStorage) Middleware

type Result

type Result struct {
	Commit     bool
	Retry      bool
	RetryError error
	RetryAfter time.Duration
}

func Commit

func Commit() Result

func Retry

func Retry(after time.Duration, err error) Result

type Sync

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

func NewSync

func NewSync(logger log.Logger, adapter SyncHandlerAdapter, middlewares ...Middleware) Sync

func (Sync) Handle

func (r Sync) Handle(ctx context.Context, delivery *consumer.Delivery)

type SyncHandlerAdapter

type SyncHandlerAdapter interface {
	Handle(ctx context.Context, delivery *consumer.Delivery) Result
}

type SyncHandlerAdapterFunc

type SyncHandlerAdapterFunc func(ctx context.Context, delivery *consumer.Delivery) Result

func (SyncHandlerAdapterFunc) Handle

func (a SyncHandlerAdapterFunc) Handle(ctx context.Context, delivery *consumer.Delivery) Result