concurr

Documentation

Index

• func ContextWithNewCancel(parent context.Context, done <-chan struct{}) context.Context
• func ContextWithTimeout(parent context.Context, clock clockwork.Clock, timeout time.Duration) (context.Context, context.CancelFunc)
• func IsPermanent(err error) bool
• func MakeChanInfCap[T any](expectedCap int) (<-chan T, chan<- T)
• func Permanent(err error) error
• func RetryWithBackoff(ctx context.Context, c clockwork.Clock, f func() error) error
• type Background
  • func NewBackground() *Background
  • func (b *Background) Close()
  • func (b *Background) Go(ctx context.Context, fn func(context.Context)) bool
• type Dedup
  • func NewDedup(w Worker) *Dedup
  • func (d *Dedup) Do(ctx context.Context, key string, r Request) error
• type DedupContr
• type Fanout
  • func NewFanout(maxConcurrent int) Fanout
  • func (o Fanout) Spawn(ctx context.Context, num int, f func(context.Context, int) error) Result
• type Request
• type Result
• type Retrier
  • func RetryOptions(initial, max time.Duration, c clockwork.Clock) Retrier
  • func (r Retrier) Retry(ctx context.Context, fn func() error) error
• type Worker

Functions

func ContextWithNewCancel

func ContextWithNewCancel(parent context.Context, done <-chan struct{}) context.Context

ContextWithNewCancel replaces the parent cancellation (if any) with the given one, while preserving the embedded values.

func ContextWithTimeout

func ContextWithTimeout(
	parent context.Context,
	clock clockwork.Clock,
	timeout time.Duration,
) (context.Context, context.CancelFunc)

func IsPermanent

func IsPermanent(err error) bool

func MakeChanInfCap

func MakeChanInfCap[T any](expectedCap int) (<-chan T, chan<- T)

MakeChainInfCap simulates a channel with infinite capacity. This will never block the sender and grow an internal buffer if necessary.

Close the input channel when finished, to free resources.

func Permanent

func Permanent(err error) error

func RetryWithBackoff

func RetryWithBackoff(ctx context.Context, c clockwork.Clock, f func() error) error

Types

type Background

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

func NewBackground

func NewBackground() *Background

func (*Background) Close

func (b *Background) Close()

func (*Background) Go

func (b *Background) Go(ctx context.Context, fn func(context.Context)) bool

type Dedup

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

func NewDedup

func NewDedup(w Worker) *Dedup

func (*Dedup) Do

func (d *Dedup) Do(ctx context.Context, key string, r Request) error

type DedupContr

type DedupContr interface {
	Request(key string) Request
	OnNextDo(key string) <-chan struct{}
}

type Fanout

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

func NewFanout

func NewFanout(maxConcurrent int) Fanout

func (Fanout) Spawn

func (o Fanout) Spawn(ctx context.Context, num int, f func(context.Context, int) error) Result

type Request

type Request interface {
	Merge(other Request) (Request, bool)
	CanReorder() bool
}

type Result

type Result interface {
	Wait() error
}

type Retrier

type Retrier struct {
	backoff.BackOff
	// contains filtered or unexported fields
}

func RetryOptions

func RetryOptions(initial, max time.Duration, c clockwork.Clock) Retrier

func (Retrier) Retry

func (r Retrier) Retry(ctx context.Context, fn func() error) error

type Worker

type Worker interface {
	Work(ctx context.Context, key string, cntr DedupContr) error
}