queueutils

Documentation

Index

• func BatchConcurrent[T any](batchSize int, things []T, fn func(group []T) error) error
• func BatchLinear[T any](batchSize int, things []T, fn func(group []T) error) error
• func SleepWithExponentialBackoff(base, max time.Duration, retryCount int)
• type ID
• type OpMethod
• type OperationPool
  • func NewOperationPool[T ID](ql *zerolog.Logger, timeout time.Duration, description string, ...) *OperationPool[T]
  • func (p *OperationPool[T]) GetOperation(id T) *SerialOperation[T]
  • func (p *OperationPool[T]) RunOrContinue(id T)
  • func (p *OperationPool[T]) SetPartitions(partitions []int64)
  • func (p *OperationPool[T]) SetTenants(tenants []*dbsqlc.Tenant)
  • func (p *OperationPool[T]) WithJitter(maxJitter time.Duration) *OperationPool[T]
• type SerialOperation
  • func (o *SerialOperation[T]) Run(ql *zerolog.Logger)
  • func (o *SerialOperation[T]) RunOrContinue(ql *zerolog.Logger)

Functions

func BatchConcurrent

func BatchConcurrent[T any](batchSize int, things []T, fn func(group []T) error) error

func BatchLinear

func BatchLinear[T any](batchSize int, things []T, fn func(group []T) error) error

func SleepWithExponentialBackoff

func SleepWithExponentialBackoff(base, max time.Duration, retryCount int)

SleepWithExponentialBackoff sleeps for a duration calculated using exponential backoff and jitter, based on the retry count. The base sleep time and maximum sleep time are provided as inputs. retryCount determines the exponential backoff multiplier.

Types

type ID

type ID interface {
	string | int64
}

type OpMethod

type OpMethod[T ID] func(ctx context.Context, id T) (bool, error)

type OperationPool

type OperationPool[T ID] struct {
	// contains filtered or unexported fields
}

func NewOperationPool

func NewOperationPool[T ID](ql *zerolog.Logger, timeout time.Duration, description string, method OpMethod[T]) *OperationPool[T]

func (*OperationPool[T]) GetOperation

func (p *OperationPool[T]) GetOperation(id T) *SerialOperation[T]

func (*OperationPool[T]) RunOrContinue

func (p *OperationPool[T]) RunOrContinue(id T)

func (*OperationPool[T]) SetPartitions

func (p *OperationPool[T]) SetPartitions(partitions []int64)

func (*OperationPool[T]) SetTenants

func (p *OperationPool[T]) SetTenants(tenants []*dbsqlc.Tenant)

func (*OperationPool[T]) WithJitter

func (p *OperationPool[T]) WithJitter(maxJitter time.Duration) *OperationPool[T]

type SerialOperation

type SerialOperation[T ID] struct {
	// contains filtered or unexported fields
}

SerialOperation represents a method that can only run serially. It can be configured with a maxJitter duration to add a random delay before executing, which helps prevent the "thundering herd" problem when many operations might start at the same time. The jitter is disabled by default (maxJitter=0) and can be enabled via OperationPool.WithJitter().

func (*SerialOperation[T]) Run

func (o *SerialOperation[T]) Run(ql *zerolog.Logger)

func (*SerialOperation[T]) RunOrContinue

func (o *SerialOperation[T]) RunOrContinue(ql *zerolog.Logger)