patterns

Documentation

Index

• Variables
• func AggregateChannel[T any](ctx context.Context, inputStream <-chan T, f AggregateFunc[T]) <-chan T
• func Detach(ctx context.Context) (context.Context, error)
• func FilterChannel[T any](ctx context.Context, inputStream <-chan T, f FilterFunc[T]) <-chan T
• func Funnel[T any](sources ...<-chan T) <-chan T
• func Pipeline[T any](jobs ...Job[T])
• func Split[T any](source <-chan T, n int) []<-chan T
• func Workers[T any](ctx context.Context, f WorkFunc[T], n int, wg *sync.WaitGroup, source <-chan T, ...)
• func Wrap(ctx context.Context, f WrappedFunc) error
• type AggregateFunc
• type Broadcast
  • func NewBroadcast[T any](ctx context.Context, source <-chan T) *Broadcast[T]
  • func (b *Broadcast[T]) CancelSubscription(channel <-chan T)
  • func (b *Broadcast[T]) Subscribe() <-chan T
• type Circuit
  • func Breaker[T any](circuit Circuit[T], waitDuration time.Duration, failureThreshold uint) Circuit[T]
  • func Debounce[T any](circuit Circuit[T], d time.Duration) Circuit[T]
• type Effector
  • func Retry[T any](effector Effector[T], retries int, delay time.Duration) Effector[T]
  • func Throttle[T any](e Effector[T], max, refill uint, d time.Duration) Effector[T]
• type FilterFunc
• type Iterator
  • func NewIterator[T any](data []T) *Iterator[T]
  • func (i *Iterator[T]) GetNext() T
  • func (i *Iterator[T]) HasNext() bool
• type Job
• type WorkFunc
• type WorkersCurator
  • func NewWorkersCurator[T any](ctx context.Context, f WorkFunc[T], source <-chan T, workerCount int) *WorkersCurator[T]
  • func (c *WorkersCurator[T]) GetErrCh() chan error
  • func (c *WorkersCurator[T]) Wait()
• type WrappedFunc

Variables

var (
	ErrCircuitWait = errors.New("err need more time wait before request")
)

var (
	ErrEmptyParentContext = errors.New("empty parent context")
)

var (
	ErrToManyCalls = errors.New("too many calls")
)

Functions

func AggregateChannel

func AggregateChannel[T any](ctx context.Context, inputStream <-chan T, f AggregateFunc[T]) <-chan T

func Detach

func Detach(ctx context.Context) (context.Context, error)

func FilterChannel

func FilterChannel[T any](ctx context.Context, inputStream <-chan T, f FilterFunc[T]) <-chan T

func Funnel

func Funnel[T any](sources ...<-chan T) <-chan T

func Pipeline

func Pipeline[T any](jobs ...Job[T])

func Split

func Split[T any](source <-chan T, n int) []<-chan T

func Workers

func Workers[T any](
	ctx context.Context,
	f WorkFunc[T],
	n int,
	wg *sync.WaitGroup,
	source <-chan T,
	errCh chan error,
	panicCh chan struct{},
)

Workers Два последних параметра опциональны, если хочется больше мониторинга. Можно переслать nil

func Wrap

func Wrap(ctx context.Context, f WrappedFunc) error

Types

type AggregateFunc

type AggregateFunc[T any] func(data T) T

type Broadcast

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

func NewBroadcast

func NewBroadcast[T any](ctx context.Context, source <-chan T) *Broadcast[T]

func (*Broadcast[T]) CancelSubscription

func (b *Broadcast[T]) CancelSubscription(channel <-chan T)

func (*Broadcast[T]) Subscribe

func (b *Broadcast[T]) Subscribe() <-chan T

type Circuit

type Circuit[T any] func(ctx context.Context) (T, error)

func Breaker

func Breaker[T any](circuit Circuit[T], waitDuration time.Duration, failureThreshold uint) Circuit[T]

func Debounce

func Debounce[T any](circuit Circuit[T], d time.Duration) Circuit[T]

type Effector

type Effector[T any] func(ctx context.Context) (T, error)

func Retry

func Retry[T any](effector Effector[T], retries int, delay time.Duration) Effector[T]

func Throttle

func Throttle[T any](e Effector[T], max, refill uint, d time.Duration) Effector[T]

type FilterFunc

type FilterFunc[T any] func(data T) bool

type Iterator

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

func NewIterator

func NewIterator[T any](data []T) *Iterator[T]

func (*Iterator[T]) GetNext

func (i *Iterator[T]) GetNext() T

func (*Iterator[T]) HasNext

func (i *Iterator[T]) HasNext() bool

type Job

type Job[T any] func(in, out chan T)

type WorkFunc

type WorkFunc[T any] func(data T) error

type WorkersCurator

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

func NewWorkersCurator

func NewWorkersCurator[T any](
	ctx context.Context,
	f WorkFunc[T],
	source <-chan T,
	workerCount int,
) *WorkersCurator[T]

func (*WorkersCurator[T]) GetErrCh

func (c *WorkersCurator[T]) GetErrCh() chan error

func (*WorkersCurator[T]) Wait

func (c *WorkersCurator[T]) Wait()

type WrappedFunc

type WrappedFunc func() error