batch_handler

Oct 8, 2025

Changes in this version

+ type BatchItem struct

+ Context context.Context

+ Delivery *consumer.Delivery

+ Result Result

+ func (b *BatchItem) Ack()

+ func (b *BatchItem) MoveToDlq(err error)

+ func (b *BatchItem) Retry(err error)

+ type BatchItems []*BatchItem

+ func (bs BatchItems) AckAll()

+ func (bs BatchItems) MoveToDlqAll(err error)

+ func (bs BatchItems) RetryAll(err error)

+ type ConsumerMetricStorage interface

+ IncDlqCount func(exchange string, routingKey string)

+ IncRetryCount func(exchange string, routingKey string)

+ IncSuccessCount func(exchange string, routingKey string)

+ ObserveConsumeDuration func(exchange string, routingKey string, t time.Duration)

+ ObserveConsumeMsgSize func(exchange string, routingKey string, size int)

+ type Handler struct

+ func New(adapter SyncHandlerAdapter, purgeInterval time.Duration, maxSize int) *Handler

+ func (r *Handler) Close()

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

+ type Middleware func(next SyncHandlerAdapter) SyncHandlerAdapter

+ func Log(logger log.Logger) Middleware

+ func Metrics(metricStorage ConsumerMetricStorage) Middleware

+ func Recovery(logger log.Logger) Middleware

+ type Result struct

+ Ack bool

+ Err error

+ MoveToDlq bool

+ Retry bool

+ type Sync struct

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

+ func (r Sync) Handle(batch BatchItems)

+ type SyncHandlerAdapter interface

+ Handle func(batch BatchItems)

+ type SyncHandlerAdapterFunc func(batch BatchItems)

+ func (a SyncHandlerAdapterFunc) Handle(batch BatchItems)