flx

package module

Go to main page

Versions in this module

v0

v0.1.9

Apr 10, 2026

Changes in this version

+ type RetryEvent struct

+ Attempt int

+ Err error

+ MaxAttempts int

+ MaxRetries int

+ NextAttempt int

+ NextDelay time.Duration

+ Retry int

type RetryOption

+ func WithOnRetry(fn func(RetryEvent)) RetryOption

+ type TimeoutLatePanicEvent struct

+ Panic any

+ Stack string

type TimeoutOption

+ func WithTimeoutLatePanicObserver(fn func(TimeoutLatePanicEvent)) TimeoutOption

v0.1.8

Apr 8, 2026

v0.1.7

Apr 8, 2026

v0.1.6

Apr 7, 2026

v0.1.5

Apr 6, 2026

Changes in this version

type Stream

+ func FlatStageErr[I, O any](ctx context.Context, in Stream[I], fn func(context.Context, I, chan<- O) error, ...) Stream[O]

+ func FlatStage[I, O any](ctx context.Context, in Stream[I], fn func(context.Context, I, chan<- O), ...) Stream[O]

+ func StageErr[I, O any](ctx context.Context, in Stream[I], fn func(context.Context, I) (O, error), ...) Stream[O]

+ func Stage[I, O any](ctx context.Context, in Stream[I], fn func(context.Context, I) O, ...) Stream[O]

+ func Tap[T any](ctx context.Context, in Stream[T], fn func(context.Context, T) error, ...) Stream[T]

+ func (s Stream[T]) Tap(ctx context.Context, fn func(context.Context, T) error, opts ...Option) Stream[T]

+ func (s Stream[T]) Through(ctx context.Context, fn func(context.Context, T) T, opts ...Option) Stream[T]

+ func (s Stream[T]) ThroughErr(ctx context.Context, fn func(context.Context, T) (T, error), opts ...Option) Stream[T]

v0.1.4

Mar 28, 2026

v0.1.3

Mar 27, 2026

v0.1.2

Mar 26, 2026

Changes in this version

+ var ErrInvalidWindowCount = errors.New("flx: window count must be greater than 0")

+ var ErrInvalidWindowDuration = errors.New("flx: window duration must be positive")

+ type Group struct

+ Items []T

+ Key K

type Stream

+ func DistinctByCount[T any, K comparable](s Stream[T], n int, fn func(T) K) Stream[T]

+ func DistinctByWindow[T any, K comparable](ctx context.Context, s Stream[T], every time.Duration, fn func(T) K) Stream[T]

+ func GroupByCount[T any, K comparable](s Stream[T], n int, fn func(T) K) Stream[Group[K, T]]

+ func GroupByWindow[T any, K comparable](ctx context.Context, s Stream[T], every time.Duration, fn func(T) K) Stream[Group[K, T]]

v0.1.1

Mar 22, 2026

v0.1.0

Mar 22, 2026

Changes in this version

+ var ErrAttemptTimeoutRequiresRetryCtx = errors.New("flx: WithAttemptTimeout requires DoWithRetryCtx")

+ var ErrCanceled = context.Canceled

+ var ErrInterruptibleWorkersRequireContextTransform = errors.New(...)

+ var ErrInvalidErrorStrategy = errors.New("flx: invalid error strategy")

+ var ErrInvalidRetryTimes = errors.New("flx: retry times must be greater than 0")

+ var ErrNegativeAttemptTimeout = errors.New("flx: attempt timeout must not be negative")

+ var ErrNegativeRetryInterval = errors.New("flx: retry interval must not be negative")

+ var ErrNegativeRetryTimeout = errors.New("flx: retry timeout must not be negative")

+ var ErrNegativeTimeout = errors.New("flx: timeout must not be negative")

+ var ErrNilContext = errors.New("flx: nil context")

+ var ErrNilController = errors.New("flx: nil concurrency controller")

+ var ErrRetryAttemptTimeout = errors.New("flx: retry attempt timeout")

+ var ErrTimeout = context.DeadlineExceeded

+ var ErrWorkerLimitReduced = errors.New("flx: worker canceled because concurrency limit was reduced")

+ func DoWithRetry(fn func() error, opts ...RetryOption) error

+ func DoWithRetryCtx(ctx context.Context, fn func(context.Context, int) error, opts ...RetryOption) error

+ func DoWithTimeout(fn func() error, timeout time.Duration, opts ...TimeoutOption) error

+ func DoWithTimeoutCtx(fn func(context.Context) error, timeout time.Duration, opts ...TimeoutOption) error

+ func Parallel(fns ...func())

+ func ParallelErr(fns ...func() error) error

+ func ParallelWithErrorStrategy(strategy ErrorStrategy, fns ...func()) error

+ func Reduce[T, R any](s Stream[T], fn func(<-chan T) (R, error)) (R, error)

+ func SendContext[T any](ctx context.Context, pipe chan<- T, item T) bool

+ type ConcurrencyController struct

+ func NewConcurrencyController(workers int) *ConcurrencyController

+ func (c *ConcurrencyController) ActiveWorkers() int

+ func (c *ConcurrencyController) SetWorkers(n int)

+ func (c *ConcurrencyController) Workers() int

+ type DynamicSemaphore struct

+ func NewDynamicSemaphore(n int) *DynamicSemaphore

+ func (s *DynamicSemaphore) Acquire()

+ func (s *DynamicSemaphore) AcquireCtx(ctx context.Context) error

+ func (s *DynamicSemaphore) Cap() int

+ func (s *DynamicSemaphore) Current() int

+ func (s *DynamicSemaphore) Release()

+ func (s *DynamicSemaphore) Resize(n int)

+ type ErrorStrategy uint8

+ const ErrorStrategyCollect

+ const ErrorStrategyFailFast

+ const ErrorStrategyLogAndContinue

+ func (s ErrorStrategy) String() string

+ type Option func(*opOptions)

+ func WithDynamicWorkers(controller *ConcurrencyController) Option

+ func WithErrorStrategy(strategy ErrorStrategy) Option

+ func WithForcedDynamicWorkers(controller *ConcurrencyController) Option

+ func WithInterruptibleWorkers(controller *ConcurrencyController) Option

+ func WithUnlimitedWorkers() Option

+ func WithWorkers(workers int) Option

+ type RetryOption func(*retryOptions)

+ func WithAttemptTimeout(timeout time.Duration) RetryOption

+ func WithIgnoreErrors(ignoreErrors []error) RetryOption

+ func WithInterval(interval time.Duration) RetryOption

+ func WithRetry(times int) RetryOption

+ func WithTimeout(timeout time.Duration) RetryOption

+ type Stream struct

+ func Chunk[T any](s Stream[T], n int) Stream[[]T]

+ func Concat[T any](s Stream[T], others ...Stream[T]) Stream[T]

+ func DistinctBy[T any, K comparable](s Stream[T], fn func(T) K) Stream[T]

+ func FlatMapContextErr[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T, chan<- U) error, ...) Stream[U]

+ func FlatMapContext[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T, chan<- U), ...) Stream[U]

+ func FlatMapErr[T, U any](s Stream[T], fn func(T, chan<- U) error, opts ...Option) Stream[U]

+ func FlatMap[T, U any](s Stream[T], fn func(T, chan<- U), opts ...Option) Stream[U]

+ func FromChan[T any](source <-chan T) Stream[T]

+ func From[T any](generate func(chan<- T)) Stream[T]

+ func GroupBy[T any, K comparable](s Stream[T], fn func(T) K) Stream[[]T]

+ func MapContextErr[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T) (U, error), ...) Stream[U]

+ func MapContext[T, U any](ctx context.Context, s Stream[T], fn func(context.Context, T) U, ...) Stream[U]

+ func MapErr[T, U any](s Stream[T], fn func(T) (U, error), opts ...Option) Stream[U]

+ func Map[T, U any](s Stream[T], fn func(T) U, opts ...Option) Stream[U]

+ func Values[T any](items ...T) Stream[T]

+ func (s Stream[T]) AllMatch(predicate func(T) bool) bool

+ func (s Stream[T]) AllMatchErr(predicate func(T) bool) (bool, error)

+ func (s Stream[T]) AnyMatch(predicate func(T) bool) bool

+ func (s Stream[T]) AnyMatchErr(predicate func(T) bool) (bool, error)

+ func (s Stream[T]) Buffer(n int) Stream[T]

+ func (s Stream[T]) Collect() []T

+ func (s Stream[T]) CollectErr() ([]T, error)

+ func (s Stream[T]) Concat(others ...Stream[T]) Stream[T]

+ func (s Stream[T]) Count() (count int)

+ func (s Stream[T]) CountErr() (int, error)

+ func (s Stream[T]) Done()

+ func (s Stream[T]) DoneErr() error

+ func (s Stream[T]) Err() error

+ func (s Stream[T]) Filter(fn func(T) bool, opts ...Option) Stream[T]

+ func (s Stream[T]) First() (T, bool)

+ func (s Stream[T]) FirstErr() (T, bool, error)

+ func (s Stream[T]) ForAll(fn func(<-chan T))

+ func (s Stream[T]) ForAllErr(fn func(<-chan T)) error

+ func (s Stream[T]) ForEach(fn func(T))

+ func (s Stream[T]) ForEachErr(fn func(T)) error

+ func (s Stream[T]) Head(n int64) Stream[T]

+ func (s Stream[T]) Last() (T, bool)

+ func (s Stream[T]) LastErr() (T, bool, error)

+ func (s Stream[T]) Max(less func(T, T) bool) (T, bool)

+ func (s Stream[T]) MaxErr(less func(T, T) bool) (T, bool, error)

+ func (s Stream[T]) Min(less func(T, T) bool) (T, bool)

+ func (s Stream[T]) MinErr(less func(T, T) bool) (T, bool, error)

+ func (s Stream[T]) NoneMatch(predicate func(T) bool) bool

+ func (s Stream[T]) NoneMatchErr(predicate func(T) bool) (bool, error)

+ func (s Stream[T]) Parallel(fn func(T), opts ...Option)

+ func (s Stream[T]) ParallelErr(fn func(T) error, opts ...Option) error

+ func (s Stream[T]) Reverse() Stream[T]

+ func (s Stream[T]) Skip(n int64) Stream[T]

+ func (s Stream[T]) Sort(less func(T, T) bool) Stream[T]

+ func (s Stream[T]) Tail(n int64) Stream[T]

+ type TimeoutOption func() context.Context

+ func WithContext(ctx context.Context) TimeoutOption

+ type WorkerError struct

+ Err error

+ func (e *WorkerError) Error() string

+ func (e *WorkerError) Unwrap() error

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL