autoretry

Documentation

Index

• Variables
• type AckFunc
• type List
  • func NewList[T any](reader ReadFunc[T], mutator MutatorFunc[T]) *List[T]
  • func (l *List[T]) Close(ctx context.Context) error
  • func (l *List[T]) Exhausted() bool
  • func (l *List[T]) Shift(ctx context.Context, enableRead bool) (t T, fn AckFunc, err error)
• type MutatorFunc
• type ReadFunc

Variables

var ErrExhausted = errors.New("retry list is exhausted")

ErrExhausted is returned by shift calls when there are no pending messages and new reads are disabled.

Types

type AckFunc

type AckFunc func(context.Context, error) error

AckFunc is a synchronous function that matches the standard acknowledgment signature in Benthos.

type List

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

List contains a slice of items that are pending an acknowledgement, once items are added it's required that all rejected adopted T values are recirculated either via TryShift (non-blocking) or Shift.

func NewList

func NewList[T any](reader ReadFunc[T], mutator MutatorFunc[T]) *List[T]

NewList returns a new list of Ts requiring automatic retries.

func (*List[T]) Close

func (l *List[T]) Close(ctx context.Context) error

Close any pending read attempts that could be dangling from prior shifts.

func (*List[T]) Exhausted

func (l *List[T]) Exhausted() bool

Exhausted returns true if all adopted Ts have been acknowledged.

func (*List[T]) Shift

func (l *List[T]) Shift(ctx context.Context, enableRead bool) (t T, fn AckFunc, err error)

Shift blocks until either a T needs retrying and returns it, enableRead is set true and a new T is ready, or the context is cancelled. Returns ErrExhausted if all messages are exhausted and enableRead is set to false.

type MutatorFunc

type MutatorFunc[T any] func(t T, err error) T

MutatorFunc is an optional closure used to mutate a T about to be scheduled for retry based on the returned error. This is useful for reducing a batch based on a batch error, etc.

type ReadFunc

type ReadFunc[T any] func(context.Context) (t T, aFn AckFunc, err error)

ReadFunc is a closure used to obtain a new T, this is done asynchronously from retries but the result is given lower priority than retries. If the read func returns an error then it is returned as a highest priority.