fun

Dec 9, 2025

Changes in this version

+ func ConvertFn[T any, O any](op fn.Converter[T, O]) interface

Oct 27, 2025

Changes in this version

+ func Convert[T any, O any](op fnx.Converter[T, O]) interface

type ChanReceive

+ func (ro ChanReceive[T]) ReadAll(op func(context.Context, T) error) fnx.Worker

type ChanSend

+ func (sm ChanSend[T]) WriteAll(iter *Stream[T]) fnx.Worker

type Constructors

+ func (Constructors) ContextChannelWorker(ctx context.Context) fnx.Worker

+ func (Constructors) ConvertOperationToWorker() fnx.Converter[fnx.Operation, fnx.Worker]

+ func (Constructors) ConvertWorkerToOperation(eh fn.Handler[error]) fnx.Converter[fnx.Worker, fnx.Operation]

+ func (Constructors) OperationHandler() fnx.Handler[fnx.Operation]

+ func (Constructors) RunAllOperations(st *Stream[fnx.Operation]) fnx.Worker

+ func (Constructors) RunAllWorkers(st *Stream[fnx.Worker]) fnx.Worker

+ func (Constructors) WorkerHandler() fnx.Handler[fnx.Worker]

type OptionProvider

+ func WorkerGroupConfDefaults() OptionProvider[*WorkerGroupConf]

May 2, 2025

Changes in this version

+ const ErrStreamContinue

+ var MAKE = Constructors

type ChanOp

+ func (op ChanOp[T]) Generator() Generator[T]

+ func (op ChanOp[T]) Handler() Handler[T]

+ func (op ChanOp[T]) Stream() *Stream[T]

type ChanReceive

+ func (ro ChanReceive[T]) Generator() Generator[T]

+ func (ro ChanReceive[T]) Handle(op Handler[T]) Worker

+ func (ro ChanReceive[T]) IteratorIndexed(ctx context.Context) iter.Seq2[int, T]

+ func (ro ChanReceive[T]) Stream() *Stream[T]

type ChanSend

+ func (sm ChanSend[T]) Handler() Handler[T]

+ type Constructors struct

+ func (Constructors) Atoi() Converter[string, int]

+ func (Constructors) ConvertErrorsToStrings() Converter[[]error, []string]

+ func (Constructors) Counter(maxVal int) *Stream[int]

+ func (Constructors) ErrorChannelWorker(ch <-chan error) Worker

+ func (Constructors) ErrorHandler(of fn.Handler[error]) fn.Handler[error]

+ func (Constructors) ErrorHandlerWithAbort(cancel context.CancelFunc) fn.Handler[error]

+ func (Constructors) ErrorHandlerWithoutCancelation(of fn.Handler[error]) fn.Handler[error]

+ func (Constructors) ErrorHandlerWithoutTerminating(of fn.Handler[error]) fn.Handler[error]

+ func (Constructors) ErrorStream(ec *erc.Collector) *Stream[error]

+ func (Constructors) ErrorUnwindTransformer(filter erc.Filter) Converter[error, []error]

+ func (Constructors) Itoa() Converter[int, string]

+ func (Constructors) Lines(reader io.Reader) *Stream[string]

+ func (Constructors) LinesWithSpaceTrimed(reader io.Reader) *Stream[string]

+ func (Constructors) OperationPool(iter *Stream[Operation]) Operation

+ func (Constructors) ProcessOperation() Handler[Operation]

+ func (Constructors) Recover(ob fn.Handler[error])

+ func (Constructors) Signal() (func(), Worker)

+ func (Constructors) Sprint(args ...any) fn.Future[string]

+ func (Constructors) Sprintf(tmpl string, args ...any) fn.Future[string]

+ func (Constructors) Sprintln(args ...any) fn.Future[string]

+ func (Constructors) Str(args []any) fn.Future[string]

+ func (Constructors) StrConcatinate(strs ...string) fn.Future[string]

+ func (Constructors) StrJoin(args []any) fn.Future[string]

+ func (Constructors) StrSliceConcatinate(input []string) fn.Future[string]

+ func (Constructors) Strf(tmpl string, args []any) fn.Future[string]

+ func (Constructors) Stringer(op fmt.Stringer) fn.Future[string]

+ func (Constructors) StringsJoin(strs []string, sep string) fn.Future[string]

+ func (Constructors) Strln(args []any) fn.Future[string]

+ func (Constructors) WorkerPool(iter *Stream[Worker]) Worker

type Converter

+ func MakeConverterErr[T any, O any](op func(T) (O, error)) Converter[T, O]

+ func MakeConverter[T any, O any](op func(T) O) Converter[T, O]

+ func MakeCovnerterOk[T any, O any](op func(T) (O, bool)) Converter[T, O]

+ func (mpf Converter[T, O]) Convert(ctx context.Context, in T) (O, error)

+ func (mpf Converter[T, O]) Generator(prod Generator[T]) Generator[O]

+ func (mpf Converter[T, O]) Lock() Converter[T, O]

+ func (mpf Converter[T, O]) Parallel(iter *Stream[T], opts ...OptionProvider[*WorkerGroupConf]) *Stream[O]

+ func (mpf Converter[T, O]) Stream(iter *Stream[T]) *Stream[O]

+ func (mpf Converter[T, O]) Wait(in T) (O, error)

+ func (mpf Converter[T, O]) WithLock(mu *sync.Mutex) Converter[T, O]

+ func (mpf Converter[T, O]) WithRecover() Converter[T, O]

type Generator

+ func CheckedGenerator[T any](op func() (T, bool)) Generator[T]

+ func FutureGenerator[T any](f fn.Future[T]) Generator[T]

+ func MakeGenerator[T any](fn func() (T, error)) Generator[T]

+ func NewGenerator[T any](fn func(ctx context.Context) (T, error)) Generator[T]

+ func PtrGenerator[T any](fn func() *T) Generator[T]

+ func StaticGenerator[T any](val T, err error) Generator[T]

+ func ValueGenerator[T any](val T) Generator[T]

+ func (pf Generator[T]) After(ts time.Time) Generator[T]

+ func (pf Generator[T]) Capture() fn.Future[T]

+ func (pf Generator[T]) Check(ctx context.Context) (T, bool)

+ func (pf Generator[T]) Delay(d time.Duration) Generator[T]

+ func (pf Generator[T]) Filter(fl func(T) bool) Generator[T]

+ func (pf Generator[T]) Force() fn.Future[T]

+ func (pf Generator[T]) Future(ctx context.Context, ob fn.Handler[error]) fn.Future[T]

+ func (pf Generator[T]) If(cond bool) Generator[T]

+ func (pf Generator[T]) Ignore(ctx context.Context) fn.Future[T]

+ func (pf Generator[T]) Jitter(jf func() time.Duration) Generator[T]

+ func (pf Generator[T]) Join(next Generator[T]) Generator[T]

+ func (pf Generator[T]) Limit(in int) Generator[T]

+ func (pf Generator[T]) Lock() Generator[T]

+ func (pf Generator[T]) Must(ctx context.Context) fn.Future[T]

+ func (pf Generator[T]) Once() Generator[T]

+ func (pf Generator[T]) Parallel(opts ...OptionProvider[*WorkerGroupConf]) Generator[T]

+ func (pf Generator[T]) PostHook(op func()) Generator[T]

+ func (pf Generator[T]) PreHook(op Operation) Generator[T]

+ func (pf Generator[T]) Read(ctx context.Context) (T, error)

+ func (pf Generator[T]) Retry(n int) Generator[T]

+ func (pf Generator[T]) Stream() *Stream[T]

+ func (pf Generator[T]) TTL(dur time.Duration) Generator[T]

+ func (pf Generator[T]) Wait() (T, error)

+ func (pf Generator[T]) When(cond func() bool) Generator[T]

+ func (pf Generator[T]) WithCancel() (Generator[T], context.CancelFunc)

+ func (pf Generator[T]) WithErrorCheck(ef fn.Future[error]) Generator[T]

+ func (pf Generator[T]) WithErrorFilter(ef erc.Filter) Generator[T]

+ func (pf Generator[T]) WithErrorHandler(handler fn.Handler[error], resolver fn.Future[error]) Generator[T]

+ func (pf Generator[T]) WithLock(mtx *sync.Mutex) Generator[T]

+ func (pf Generator[T]) WithLocker(mtx sync.Locker) Generator[T]

+ func (pf Generator[T]) WithRecover() Generator[T]

+ func (pf Generator[T]) WithoutErrors(errs ...error) Generator[T]

type Handler

+ func FromHandler[T any](f fn.Handler[T]) Handler[T]

+ func JoinHandlers[T any](pfs ...Handler[T]) Handler[T]

+ func MakeHandler[T any](fn func(T) error) Handler[T]

+ func NewHandler[T any](fn func(context.Context, T) error) Handler[T]

+ func (pf Handler[T]) After(ts time.Time) Handler[T]

+ func (pf Handler[T]) Delay(dur time.Duration) Handler[T]

+ func (pf Handler[T]) Force(in T)

+ func (pf Handler[T]) Handler(ctx context.Context, oe fn.Handler[error]) fn.Handler[T]

+ func (pf Handler[T]) Ignore(ctx context.Context, in T)

+ func (pf Handler[T]) Jitter(jf func() time.Duration) Handler[T]

+ func (pf Handler[T]) Limit(n int) Handler[T]

+ func (pf Handler[T]) Must(ctx context.Context, in T)

+ func (pf Handler[T]) PostHook(op func()) Handler[T]

+ func (pf Handler[T]) Read(ctx context.Context, in T) error

+ func (pf Handler[T]) ReadAll(st *Stream[T]) Worker

+ func (pf Handler[T]) TTL(dur time.Duration) Handler[T]

+ func (pf Handler[T]) Wait(in T) error

+ func (pf Handler[T]) WithCancel() (Handler[T], context.CancelFunc)

+ func (pf Handler[T]) WithErrorCheck(ef fn.Future[error]) Handler[T]

+ func (pf Handler[T]) WithErrorFilter(ef erc.Filter) Handler[T]

+ func (pf Handler[T]) WithLocker(mtx sync.Locker) Handler[T]

+ func (pf Handler[T]) WithoutErrors(errs ...error) Handler[T]

type Operation

+ func (wf Operation) Group(n int) Operation

+ func (wf Operation) WithErrorHook(ef fn.Future[error]) Worker

type RuntimeInvariant

+ func (RuntimeInvariant) New(args ...any) error

+ type Stream struct

+ func ChannelStream[T any](ch <-chan T) *Stream[T]

+ func JoinStreams[T any](iters ...*Stream[T]) *Stream[T]

+ func MakeStream[T any](gen Generator[T]) *Stream[T]

+ func MergeStreams[T any](iters *Stream[*Stream[T]]) *Stream[T]

+ func SeqStream[T any](it iter.Seq[T]) *Stream[T]

+ func SliceStream[T any](in []T) *Stream[T]

+ func VariadicStream[T any](in ...T) *Stream[T]

+ func (st *Stream[T]) AddError(e error)

+ func (st *Stream[T]) Any() *Stream[any]

+ func (st *Stream[T]) Buffer(n int) *Stream[T]

+ func (st *Stream[T]) BufferParallel(n int) *Stream[T]

+ func (st *Stream[T]) BufferedChannel(ctx context.Context, size int) <-chan T

+ func (st *Stream[T]) Channel(ctx context.Context) <-chan T

+ func (st *Stream[T]) Close() error

+ func (st *Stream[T]) CloseHook() func(*Stream[T])

+ func (st *Stream[T]) Count(ctx context.Context) (count int)

+ func (st *Stream[T]) ErrorHandler() fn.Handler[error]

+ func (st *Stream[T]) Filter(check func(T) bool) *Stream[T]

+ func (st *Stream[T]) Generator() Generator[T]

+ func (st *Stream[T]) Iterator(ctx context.Context) iter.Seq[T]

+ func (st *Stream[T]) Join(iters ...*Stream[T]) *Stream[T]

+ func (st *Stream[T]) MarshalJSON() ([]byte, error)

+ func (st *Stream[T]) Next(ctx context.Context) bool

+ func (st *Stream[T]) Parallel(fn Handler[T], opts ...OptionProvider[*WorkerGroupConf]) Worker

+ func (st *Stream[T]) Read(ctx context.Context) (out T, err error)

+ func (st *Stream[T]) ReadAll(fn fn.Handler[T]) Worker

+ func (st *Stream[T]) Reduce(reducer func(T, T) (T, error)) Generator[T]

+ func (st *Stream[T]) Slice(ctx context.Context) (out []T, _ error)

+ func (st *Stream[T]) Split(num int) []*Stream[T]

+ func (st *Stream[T]) Transform(op Converter[T, T]) *Stream[T]

+ func (st *Stream[T]) UnmarshalJSON(in []byte) error

+ func (st *Stream[T]) Value() T

+ func (st *Stream[T]) WithHook(hook fn.Handler[*Stream[T]]) *Stream[T]

type WaitGroup

+ func (wg *WaitGroup) Group(n int, op Operation) Operation

+ func (wg *WaitGroup) StartGroup(ctx context.Context, n int, op Operation) Operation

type Worker

+ func (wf Worker) WithErrorHook(ef fn.Future[error]) Worker

type WorkerGroupConf

+ ErrorCollector *erc.Collector

Jan 21, 2025

Changes in this version

type ChanOp

+ func (op ChanOp[T]) Seq(ctx context.Context) iter.Seq[T]

+ func (op ChanOp[T]) Seq2(ctx context.Context) iter.Seq2[int, T]

type ChanReceive

+ func (ro ChanReceive[T]) Seq(ctx context.Context) iter.Seq[T]

+ func (ro ChanReceive[T]) Seq2(ctx context.Context) iter.Seq2[int, T]

type Handlers

+ func (Handlers) ErrorStackIterator(s *ers.Stack) *Iterator[error]

+ func (Handlers) ErrorStackProducer(s *ers.Stack) Producer[error]

type Iterator

+ func (i *Iterator[T]) Seq(ctx context.Context) iter.Seq[T]

Jan 14, 2024

Changes in this version

type Handlers

+ func (Handlers) LinesWithSpaceTrimed(reader io.Reader) *Iterator[string]

type RuntimeInvariant

+ func (RuntimeInvariant) Ok(cond bool, args ...any)

type Transform

+ func ConverterOk[T any, O any](op func(T) (O, bool)) Transform[T, O]

Oct 29, 2023

Changes in this version

type Handlers

+ func (Handlers) WorkerPool(iter *Iterator[Worker]) Worker

type Processor

+ func ProcessorGroup(pfs ...Processor[T]) Processor[T]

type WaitGroup

+ func (wg *WaitGroup) Inc()

Aug 31, 2023

Changes in this version

type Processor

+ func MakeHandlerProcessor(fn func(T)) Processor[T]

+ func NewProcessor(fn func(context.Context, T) error) Processor[T]

type Producer

+ func NewProducer(fn func(ctx context.Context) (T, error)) Producer[T]

Aug 30, 2023

Changes in this version

type Processor

+ func (pf Processor[T]) Capture() Handler[T]

+ func (pf Processor[T]) Wait(in T) error

+ func (pf Processor[T]) WithErrorCheck(ef Future[error]) Processor[T]

type Producer

+ func (pf Producer[T]) Resolve(ctx context.Context) (T, error)

+ func (pf Producer[T]) Wait() (T, error)

+ func (pf Producer[T]) WithErrorCheck(ef Future[error]) Producer[T]

type Worker

+ func (wf Worker) WithErrorCheck(ef Future[error]) Worker

Aug 29, 2023

Changes in this version

type ChanOp

+ func (op ChanOp[T]) Cap() int

+ func (op ChanOp[T]) Len() int

type ChanReceive

+ func (ro ChanReceive[T]) Filter(ctx context.Context, eh Handler[error], filter func(T) bool) ChanReceive[T]

type Handler

+ func (of Handler[T]) PreHook(prev Handler[T]) Handler[T]

+ func (of Handler[T]) RecoverPanic(in T) error

+ func (of Handler[T]) WithRecover(oe Handler[error]) Handler[T]

type Handlers

+ func (Handlers) ErrorStackHandler() (*ers.Stack, Handler[error])

type Processor

+ func (pf Processor[T]) Filter(fl func(T) bool) Processor[T]

+ func (pf Processor[T]) WithErrorFilter(ef ers.Filter) Processor[T]

+ func (pf Processor[T]) WithRecover() Processor[T]

type Producer

+ func (pf Producer[T]) Filter(fl func(T) bool) Producer[T]

+ func (pf Producer[T]) IteratorWithErrorCollector(ec Handler[error], er Future[error]) *Iterator[T]

+ func (pf Producer[T]) WithErrorFilter(ef ers.Filter) Producer[T]

+ func (pf Producer[T]) WithRecover() Producer[T]

type Transform

+ func (mpf Transform[T, O]) ConvertFuture(fn Future[T]) Producer[O]

+ func (mpf Transform[T, O]) ConvertProducer(fn Producer[T]) Producer[O]

+ func (mpf Transform[T, O]) Process(iter *Iterator[T]) *Iterator[O]

+ func (mpf Transform[T, O]) ProcessPipe(in Producer[T], out Processor[O]) Worker

+ func (mpf Transform[T, O]) Run(ctx context.Context, in T) (O, error)

+ func (mpf Transform[T, O]) WorkerFuture(fn Future[T], hf Handler[O]) Worker

type Worker

+ func (wf Worker) WithErrorFilter(ef ers.Filter) Worker

Aug 24, 2023

Changes in this version

type ChanOp

+ func Chan(args ...int) ChanOp[T]

+ func DefaultChan(input chan T, args ...int) ChanOp[T]

+ func (op ChanOp[T]) Blocking() ChanOp[T]

+ func (op ChanOp[T]) NonBlocking() ChanOp[T]

type Future

+ func (f Future[T]) Resolve() T

type Handler

+ func (of Handler[T]) Handle(in T)

type Handlers

+ func (Handlers) Recover(ob Handler[error])

type Operation

+ func MakeOperation(in func()) Operation

+ func (wf Operation) WithRecover() Worker

type Processor

+ func (pf Processor[T]) Parallel(ops ...T) Worker

+ func (pf Processor[T]) Retry(n int, in T) Worker

type Producer

+ func CheckProducer(op func() (T, bool)) Producer[T]

+ func (pf Producer[T]) Retry(n int) Producer[T]

type Transform

+ func (mpf Transform[T, O]) CheckWait() func(T) (O, bool)

+ func (mpf Transform[T, O]) Wait() func(T) (O, error)

+ func (mpf Transform[T, O]) WithRecover() Transform[T, O]

type Worker

+ func MakeWorker(fn func() error) Worker

+ func (wf Worker) Retry(n int) Worker

+ func (wf Worker) WithRecover() Worker

Aug 14, 2023

Changes in this version

+ func HandlePassthrough(hf Handler[O]) func(T, O) T

type ChanReceive

+ func (ro ChanReceive[T]) OK() bool

type Handler

+ func (of Handler[T]) All(in ...T)

+ func (of Handler[T]) Chain(chain ...Handler[T]) Handler[T]

type Handlers

+ func (Handlers) ErrorHandlerSingle() (Handler[error], Future[error])

+ func (Handlers) ErrorHandlerWithAbort(cancel context.CancelFunc) Handler[error]

+ func (Handlers) StrSliceConcatinate(input []string) Future[string]

type Operation

+ func (wf Operation) Interval(dur time.Duration) Operation

+ func (wf Operation) Wait()

+ func (wf Operation) While() Operation

type OptionProvider

+ func (op OptionProvider[T]) Build(conf T) (out T, err error)

type Processor

+ func ProcessifyHandler(fn Handler[T]) Processor[T]

+ func (pf Processor[T]) Add(ctx context.Context, wg *WaitGroup, eh Handler[error], op T)

+ func (pf Processor[T]) Background(ctx context.Context, op T) Worker

+ func (pf Processor[T]) Group(ctx context.Context, ops ...T) Worker

type Producer

+ func MakeProducer(fn func() (T, error)) Producer[T]

+ func (pf Producer[T]) CheckForce() (T, bool)

+ func (pf Producer[T]) Launch(ctx context.Context) Producer[T]

type Worker

+ func (wf Worker) Group(n int) Worker

+ func (wf Worker) Interval(dur time.Duration) Worker

+ func (wf Worker) Launch(ctx context.Context) Worker

+ func (wf Worker) Wait() error

Jul 26, 2023

Changes in this version

+ type Handler func(T)

+ func Handle(in func(T)) Handler[T]

+ func (of Handler[T]) Capture(in T) func()

+ func (of Handler[T]) Check(in T) error

+ func (of Handler[T]) Filter(filter func(T) T) Handler[T]

+ func (of Handler[T]) If(cond bool) Handler[T]

+ func (of Handler[T]) Iterator(iter *Iterator[T]) Worker

+ func (of Handler[T]) Join(next Handler[T]) Handler[T]

+ func (of Handler[T]) Lock() Handler[T]

+ func (of Handler[T]) Once() Handler[T]

+ func (of Handler[T]) Operation(in T) Operation

+ func (of Handler[T]) Processor() Processor[T]

+ func (of Handler[T]) Safe(oe Handler[error]) Handler[T]

+ func (of Handler[T]) Skip(hook func(T) bool) Handler[T]

+ func (of Handler[T]) When(cond func() bool) Handler[T]

+ func (of Handler[T]) WithLock(mtx *sync.Mutex) Handler[T]

+ func (of Handler[T]) Worker(in T) Worker

type Handlers

+ func (Handlers) ErrorHandler(of Handler[error]) Handler[error]

+ func (Handlers) ErrorHandlerWithoutEOF(of Handler[error]) Handler[error]

+ func (Handlers) ErrorHandlerWithoutTerminating(of Handler[error]) Handler[error]

type Iterator

+ func (i *Iterator[T]) ErrorHandler() Handler[error]

type OptionProvider

+ func WorkerGroupConfErrorHandler(observer Handler[error]) OptionProvider[*WorkerGroupConf]

type Processor

+ func MakeProcessor(fn func(T) error) Processor[T]

+ func Processify(fn func(context.Context, T) error) Processor[T]

+ func (pf Processor[T]) Handler(ctx context.Context, oe Handler[error]) Handler[T]

type WorkerGroupConf

+ ErrorHandler Handler[error]

Jul 13, 2023

Changes in this version

type Future

+ func Translate(in Future[T], tfn func(T) O) Future[O]

+ func (f Future[T]) Limit(in int) Future[T]

+ func (f Future[T]) TTL(dur time.Duration) Future[T]

type Handlers

+ func (Handlers) Atoi() Transform[string, int]

+ func (Handlers) Counter(max int) *Iterator[int]

+ func (Handlers) Itoa() Transform[int, string]

+ func (Handlers) Lines(reader io.Reader) *Iterator[string]

type Iterator

+ func (i *Iterator[T]) Transform(op Transform[T, T]) *Iterator[T]

type RuntimeInvariant

+ func (RuntimeInvariant) Failure(args ...any)

Jul 7, 2023

Changes in this version

+ const ErrIteratorSkip

+ const ErrNonBlockingChannelOperationSkipped

+ type ChanOp struct

+ func (op ChanOp[T]) Channel() chan T

+ func (op ChanOp[T]) Close()

+ func (op ChanOp[T]) Iterator() *Iterator[T]

+ func (op ChanOp[T]) Pipe() (Processor[T], Producer[T])

+ func (op ChanOp[T]) Processor() Processor[T]

+ func (op ChanOp[T]) Producer() Producer[T]

+ func (op ChanOp[T]) Receive() ChanReceive[T]

+ func (op ChanOp[T]) Send() ChanSend[T]

+ type ChanReceive struct

+ func BlockingReceive(ch <-chan T) ChanReceive[T]

+ func NonBlockingReceive(ch <-chan T) ChanReceive[T]

+ func (ro ChanReceive[T]) Check(ctx context.Context) (T, bool)

+ func (ro ChanReceive[T]) Consume(op Processor[T]) Worker

+ func (ro ChanReceive[T]) Drop(ctx context.Context) bool

+ func (ro ChanReceive[T]) Force(ctx context.Context) (out T)

+ func (ro ChanReceive[T]) Ignore(ctx context.Context)

+ func (ro ChanReceive[T]) Iterator() *Iterator[T]

+ func (ro ChanReceive[T]) Ok() bool

+ func (ro ChanReceive[T]) Producer() Producer[T]

+ func (ro ChanReceive[T]) Read(ctx context.Context) (T, error)

+ type ChanSend struct

+ func BlockingSend(ch chan<- T) ChanSend[T]

+ func NonBlockingSend(ch chan<- T) ChanSend[T]

+ func (sm ChanSend[T]) Check(ctx context.Context, it T) bool

+ func (sm ChanSend[T]) Consume(iter *Iterator[T]) Worker

+ func (sm ChanSend[T]) Ignore(ctx context.Context, it T)

+ func (sm ChanSend[T]) Processor() Processor[T]

+ func (sm ChanSend[T]) Signal(ctx context.Context)

+ func (sm ChanSend[T]) Write(ctx context.Context, it T) (err error)

+ func (sm ChanSend[T]) Zero(ctx context.Context) error

+ type Future func() T

+ func AsFuture(in T) Future[T]

+ func Futurize(f func() T) Future[T]

+ func (f Future[T]) If(cond bool) Future[T]

+ func (f Future[T]) Ignore() func()

+ func (f Future[T]) Join(merge func(T, T) T, ops ...Future[T]) Future[T]

+ func (f Future[T]) Lock() Future[T]

+ func (f Future[T]) Not(cond bool) Future[T]

+ func (f Future[T]) Once() Future[T]

+ func (f Future[T]) PostHook(fn func()) Future[T]

+ func (f Future[T]) PreHook(fn func()) Future[T]

+ func (f Future[T]) Producer() Producer[T]

+ func (f Future[T]) Reduce(merge func(T, T) T, next Future[T]) Future[T]

+ func (f Future[T]) Run() T

+ func (f Future[T]) Slice() func() []T

+ func (f Future[T]) When(c func() bool) Future[T]

+ func (f Future[T]) WithLock(m *sync.Mutex) Future[T]

+ type Handlers struct

+ var HF Handlers = Handlers{}

+ func (Handlers) ErrorCollector() (ob Observer[error], prod Producer[[]error])

+ func (Handlers) ErrorObserver(of Observer[error]) Observer[error]

+ func (Handlers) ErrorObserverWithoutEOF(of Observer[error]) Observer[error]

+ func (Handlers) ErrorObserverWithoutTerminating(of Observer[error]) Observer[error]

+ func (Handlers) ErrorProcessor(pf Processor[error]) Processor[error]

+ func (Handlers) ErrorUnwindTransformer(filter ers.Filter) Transform[error, []error]

+ func (Handlers) OperationPool(iter *Iterator[Operation]) Operation

+ func (Handlers) ProcessOperation() Processor[Operation]

+ func (Handlers) ProcessWorker() Processor[Worker]

+ func (Handlers) Sprint(args ...any) Future[string]

+ func (Handlers) Sprintf(tmpl string, args ...any) Future[string]

+ func (Handlers) Sprintln(args ...any) Future[string]

+ func (Handlers) Str(args []any) Future[string]

+ func (Handlers) StrConcatinate(strs ...string) Future[string]

+ func (Handlers) StrJoin(strs []string) Future[string]

+ func (Handlers) StrJoinWith(input []string, sep string) Future[string]

+ func (Handlers) Strf(tmpl string, args []any) Future[string]

+ func (Handlers) Stringer(op fmt.Stringer) Future[string]

+ func (Handlers) Strln(args []any) Future[string]

type Iterator

+ func ChannelIterator(ch <-chan T) *Iterator[T]

+ func ConvertIterator(iter *Iterator[T], op Transform[T, O]) *Iterator[O]

+ func MergeIterators(iters ...*Iterator[T]) *Iterator[T]

+ func SliceIterator(in []T) *Iterator[T]

+ func VariadicIterator(in ...T) *Iterator[T]

+ func (i *Iterator[T]) AddError(e error)

+ func (i *Iterator[T]) Any() *Iterator[any]

+ func (i *Iterator[T]) Buffer(n int) *Iterator[T]

+ func (i *Iterator[T]) BufferedChannel(ctx context.Context, size int) <-chan T

+ func (i *Iterator[T]) Channel(ctx context.Context) <-chan T

+ func (i *Iterator[T]) Count(ctx context.Context) int

+ func (i *Iterator[T]) ErrorObserver() Observer[error]

+ func (i *Iterator[T]) Filter(check func(T) bool) *Iterator[T]

+ func (i *Iterator[T]) Join(iters ...*Iterator[T]) *Iterator[T]

+ func (i *Iterator[T]) MarshalJSON() ([]byte, error)

+ func (i *Iterator[T]) Observe(ctx context.Context, fn Observer[T]) (err error)

+ func (i *Iterator[T]) ParallelBuffer(n int) *Iterator[T]

+ func (i *Iterator[T]) Process(fn Processor[T]) Worker

+ func (i *Iterator[T]) ProcessParallel(fn Processor[T], optp ...OptionProvider[*WorkerGroupConf]) Worker

+ func (i *Iterator[T]) Producer() Producer[T]

+ func (i *Iterator[T]) ReadOne(ctx context.Context) (out T, err error)

+ func (i *Iterator[T]) Reduce(reducer func(T, T) (T, error)) Producer[T]

+ func (i *Iterator[T]) Slice(ctx context.Context) (out []T, _ error)

+ func (i *Iterator[T]) Split(num int) []*Iterator[T]

+ func (i *Iterator[T]) UnmarshalJSON(in []byte) error

type Observer

+ func (of Observer[T]) Capture(in T) func()

+ func (of Observer[T]) Check(in T) error

+ func (of Observer[T]) Filter(filter func(T) T) Observer[T]

+ func (of Observer[T]) If(cond bool) Observer[T]

+ func (of Observer[T]) Iterator(iter *Iterator[T]) Worker

+ func (of Observer[T]) Join(next Observer[T]) Observer[T]

+ func (of Observer[T]) Lock() Observer[T]

+ func (of Observer[T]) Once() Observer[T]

+ func (of Observer[T]) Operation(in T) Operation

+ func (of Observer[T]) Processor() Processor[T]

+ func (of Observer[T]) Skip(hook func(T) bool) Observer[T]

+ func (of Observer[T]) When(cond func() bool) Observer[T]

+ func (of Observer[T]) WithLock(mtx *sync.Mutex) Observer[T]

+ type Operation func(context.Context)

+ func BlockingOperation(in func()) Operation

+ func (wf Operation) Add(ctx context.Context, wg *WaitGroup)

+ func (wf Operation) After(ts time.Time) Operation

+ func (wf Operation) Background(wg *WaitGroup) Operation

+ func (wf Operation) Block()

+ func (wf Operation) Delay(dur time.Duration) Operation

+ func (wf Operation) Future(ctx context.Context) Operation

+ func (wf Operation) Go(ctx context.Context)

+ func (wf Operation) If(cond bool) Operation

+ func (wf Operation) Jitter(dur func() time.Duration) Operation

+ func (wf Operation) Join(op Operation) Operation

+ func (wf Operation) Launch() Operation

+ func (wf Operation) Limit(in int) Operation

+ func (wf Operation) Lock() Operation

+ func (wf Operation) Once() Operation

+ func (wf Operation) PostHook(hook func()) Operation

+ func (wf Operation) PreHook(hook Operation) Operation

+ func (wf Operation) Run(ctx context.Context)

+ func (wf Operation) Safe() Worker

+ func (wf Operation) Signal(ctx context.Context) <-chan struct{}

+ func (wf Operation) StartGroup(ctx context.Context, wg *WaitGroup, n int)

+ func (wf Operation) TTL(dur time.Duration) Operation

+ func (wf Operation) When(cond func() bool) Operation

+ func (wf Operation) WithCancel() (Operation, context.CancelFunc)

+ func (wf Operation) WithLock(mtx *sync.Mutex) Operation

+ func (wf Operation) Worker() Worker

+ type OptionProvider func(T) error

+ func JoinOptionProviders(op ...OptionProvider[T]) OptionProvider[T]

+ func WorkerGroupConfAddExcludeErrors(errs ...error) OptionProvider[*WorkerGroupConf]

+ func WorkerGroupConfContinueOnError() OptionProvider[*WorkerGroupConf]

+ func WorkerGroupConfContinueOnPanic() OptionProvider[*WorkerGroupConf]

+ func WorkerGroupConfErrorCollectorPair(ob Observer[error], resolver Producer[[]error]) OptionProvider[*WorkerGroupConf]

+ func WorkerGroupConfErrorObserver(observer Observer[error]) OptionProvider[*WorkerGroupConf]

+ func WorkerGroupConfErrorResolver(resolver func() error) OptionProvider[*WorkerGroupConf]

+ func WorkerGroupConfIncludeContextErrors() OptionProvider[*WorkerGroupConf]

+ func WorkerGroupConfNumWorkers(num int) OptionProvider[*WorkerGroupConf]

+ func WorkerGroupConfSet(opt *WorkerGroupConf) OptionProvider[*WorkerGroupConf]

+ func WorkerGroupConfWithErrorCollector(ec interface{ ... }) OptionProvider[*WorkerGroupConf]

+ func WorkerGroupConfWorkerPerCPU() OptionProvider[*WorkerGroupConf]

+ func (op OptionProvider[T]) Apply(in T) error

+ func (op OptionProvider[T]) Join(opps ...OptionProvider[T]) OptionProvider[T]

+ type Processor func(context.Context, T) error

+ func BlockingProcessor(fn func(T) error) Processor[T]

+ func (pf Processor[T]) After(ts time.Time) Processor[T]

+ func (pf Processor[T]) Block(in T) error

+ func (pf Processor[T]) Check(ctx context.Context, in T) bool

+ func (pf Processor[T]) Delay(dur time.Duration) Processor[T]

+ func (pf Processor[T]) FilterErrors(ef ers.Filter) Processor[T]

+ func (pf Processor[T]) Force(in T)

+ func (pf Processor[T]) Future(ctx context.Context, in T) Worker

+ func (pf Processor[T]) If(c bool) Processor[T]

+ func (pf Processor[T]) Ignore(ctx context.Context, in T)

+ func (pf Processor[T]) Iterator(iter *Iterator[T]) Worker

+ func (pf Processor[T]) Jitter(jf func() time.Duration) Processor[T]

+ func (pf Processor[T]) Join(next Processor[T]) Processor[T]

+ func (pf Processor[T]) Limit(n int) Processor[T]

+ func (pf Processor[T]) Lock() Processor[T]

+ func (pf Processor[T]) Observer(ctx context.Context, oe Observer[error]) Observer[T]

+ func (pf Processor[T]) Once() Processor[T]

+ func (pf Processor[T]) Operation(in T, of Observer[error]) Operation

+ func (pf Processor[T]) PostHook(op func()) Processor[T]

+ func (pf Processor[T]) PreHook(op Operation) Processor[T]

+ func (pf Processor[T]) ReadAll(prod Producer[T]) Worker

+ func (pf Processor[T]) ReadOne(prod Producer[T]) Worker

+ func (pf Processor[T]) Run(ctx context.Context, in T) error

+ func (pf Processor[T]) Safe() Processor[T]

+ func (pf Processor[T]) TTL(dur time.Duration) Processor[T]

+ func (pf Processor[T]) When(c func() bool) Processor[T]

+ func (pf Processor[T]) WithCancel() (Processor[T], context.CancelFunc)

+ func (pf Processor[T]) WithLock(mtx *sync.Mutex) Processor[T]

+ func (pf Processor[T]) WithoutErrors(errs ...error) Processor[T]

+ func (pf Processor[T]) Worker(in T) Worker

+ type Producer func(context.Context) (T, error)

+ func BlockingProducer(fn func() (T, error)) Producer[T]

+ func ConsistentProducer(fn func() T) Producer[T]

+ func MakeFuture(ch <-chan T) Producer[T]

+ func StaticProducer(val T, err error) Producer[T]

+ func ValueProducer(val T) Producer[T]

+ func (pf Producer[T]) After(ts time.Time) Producer[T]

+ func (pf Producer[T]) Background(ctx context.Context, of Observer[T]) Worker

+ func (pf Producer[T]) Block() (T, error)

+ func (pf Producer[T]) Check(ctx context.Context) (T, bool)

+ func (pf Producer[T]) CheckBlock() (T, bool)

+ func (pf Producer[T]) Delay(d time.Duration) Producer[T]

+ func (pf Producer[T]) FilterErrors(ef ers.Filter) Producer[T]

+ func (pf Producer[T]) Force() T

+ func (pf Producer[T]) Future(ctx context.Context, ob Observer[error]) Future[T]

+ func (pf Producer[T]) GenerateParallel(optp ...OptionProvider[*WorkerGroupConf]) *Iterator[T]

+ func (pf Producer[T]) If(cond bool) Producer[T]

+ func (pf Producer[T]) Ignore(ctx context.Context) T

+ func (pf Producer[T]) Iterator() *Iterator[T]

+ func (pf Producer[T]) IteratorWithHook(hook func(*Iterator[T])) *Iterator[T]

+ func (pf Producer[T]) Jitter(jf func() time.Duration) Producer[T]

+ func (pf Producer[T]) Join(next Producer[T]) Producer[T]

+ func (pf Producer[T]) LaunchFuture(ctx context.Context) Producer[T]

+ func (pf Producer[T]) Limit(in int) Producer[T]

+ func (pf Producer[T]) Lock() Producer[T]

+ func (pf Producer[T]) Must(ctx context.Context) T

+ func (pf Producer[T]) Once() Producer[T]

+ func (pf Producer[T]) Operation(of Observer[T], eo Observer[error]) Operation

+ func (pf Producer[T]) PostHook(op func()) Producer[T]

+ func (pf Producer[T]) PreHook(op Operation) Producer[T]

+ func (pf Producer[T]) Run(ctx context.Context) (T, error)

+ func (pf Producer[T]) Safe() Producer[T]

+ func (pf Producer[T]) SendAll(proc Processor[T]) Worker

+ func (pf Producer[T]) SendOne(proc Processor[T]) Worker

+ func (pf Producer[T]) TTL(dur time.Duration) Producer[T]

+ func (pf Producer[T]) When(cond func() bool) Producer[T]

+ func (pf Producer[T]) WithCancel() (Producer[T], context.CancelFunc)

+ func (pf Producer[T]) WithLock(mtx *sync.Mutex) Producer[T]

+ func (pf Producer[T]) WithoutErrors(errs ...error) Producer[T]

+ func (pf Producer[T]) Worker(of Observer[T]) Worker

+ type RuntimeInvariant struct

+ func (RuntimeInvariant) IsFalse(cond bool, args ...any)

+ func (RuntimeInvariant) IsTrue(cond bool, args ...any)

+ func (RuntimeInvariant) Must(err error, args ...any)

+ func (RuntimeInvariant) OK(cond bool, args ...any)

type Transform

+ func Converter(op func(T) O) Transform[T, O]

+ func ConverterErr(op func(T) (O, error)) Transform[T, O]

+ func ConverterOK(op func(T) (O, bool)) Transform[T, O]

+ func (mpf Transform[T, O]) Block() func(T) (O, error)

+ func (mpf Transform[T, O]) BlockCheck() func(T) (O, bool)

+ func (mpf Transform[T, O]) Convert(iter *Iterator[T]) *Iterator[O]

+ func (mpf Transform[T, O]) Lock() Transform[T, O]

+ func (mpf Transform[T, O]) Pipe() (in Processor[T], out Producer[O], closer func())

+ func (mpf Transform[T, O]) ProcessParallel(iter *Iterator[T], optp ...OptionProvider[*WorkerGroupConf]) *Iterator[O]

+ func (mpf Transform[T, O]) Producer(prod Producer[T]) Producer[O]

+ func (mpf Transform[T, O]) Safe() Transform[T, O]

+ func (mpf Transform[T, O]) WithLock(mu *sync.Mutex) Transform[T, O]

+ func (mpf Transform[T, O]) Worker(in Producer[T], out Processor[O]) Worker

type WaitGroup

+ func (wg *WaitGroup) DoTimes(ctx context.Context, n int, op Operation)

+ func (wg *WaitGroup) Launch(ctx context.Context, op Operation)

+ func (wg *WaitGroup) Operation() Operation

+ func (wg *WaitGroup) Worker() Worker

+ type Worker func(context.Context) error

+ func Pipe(from Producer[T], to Processor[T]) Worker

+ func WorkerFuture(ch <-chan error) Worker

+ func (wf Worker) After(ts time.Time) Worker

+ func (wf Worker) Background(ctx context.Context, ob Observer[error])

+ func (wf Worker) Block() error

+ func (wf Worker) Check(ctx context.Context) bool

+ func (wf Worker) Delay(dur time.Duration) Worker

+ func (wf Worker) FilterErrors(ef ers.Filter) Worker

+ func (wf Worker) Future(ctx context.Context) Worker

+ func (wf Worker) If(cond bool) Worker

+ func (wf Worker) Ignore() Operation

+ func (wf Worker) Jitter(jf func() time.Duration) Worker

+ func (wf Worker) Join(next Worker) Worker

+ func (wf Worker) Limit(n int) Worker

+ func (wf Worker) Lock() Worker

+ func (wf Worker) Must() Operation

+ func (wf Worker) Observe(ctx context.Context, ob Observer[error])

+ func (wf Worker) Once() Worker

+ func (wf Worker) Operation(ob Observer[error]) Operation

+ func (wf Worker) PostHook(op func()) Worker

+ func (wf Worker) PreHook(op Operation) Worker

+ func (wf Worker) Run(ctx context.Context) error

+ func (wf Worker) Safe() Worker

+ func (wf Worker) Signal(ctx context.Context) <-chan error

+ func (wf Worker) StartGroup(ctx context.Context, n int) Worker

+ func (wf Worker) TTL(dur time.Duration) Worker

+ func (wf Worker) When(cond func() bool) Worker

+ func (wf Worker) While() Worker

+ func (wf Worker) WithCancel() (Worker, context.CancelFunc)

+ func (wf Worker) WithLock(mtx *sync.Mutex) Worker

+ func (wf Worker) WithoutErrors(errs ...error) Worker

+ type WorkerGroupConf struct

+ ContinueOnError bool

+ ContinueOnPanic bool

+ ErrorObserver Observer[error]

+ ErrorResolver func() error

+ ExcludedErrors []error

+ IncludeContextExpirationErrors bool

+ NumWorkers int

+ func (o *WorkerGroupConf) Validate() error

+ func (o WorkerGroupConf) CanContinueOnError(err error) bool

May 18, 2023

Changes in this version

+ var ErrSkippedNonBlockingChannelOperation = internal.ErrSkippedNonBlockingChannelOperation

+ func Apply(fn func(T) T, in []T) []T

+ type ChannelOp struct

+ func (op ChannelOp[T]) Recieve() Receive[T]

+ func (op ChannelOp[T]) Send() Send[T]

type Iterator

+ func BuildPairs(ctx context.Context, iter Iterator[V], keyf func(V) K) Iterator[Pair[K, V]]

+ func UnwindIterator(root T) Iterator[T]

type Map

+ func (m Map[K, V]) Check(key K) bool

+ func (m Map[K, V]) Get(key K) V

+ func (m Map[K, V]) Load(key K) (V, bool)

+ func (m Map[K, V]) SetDefault(key K)

type Pairs

+ func (p *Pairs[K, V]) Extend(toAdd Pairs[K, V])

+ type Receive struct

+ func (ro Receive[T]) Check(ctx context.Context) (T, bool)

+ func (ro Receive[T]) Drop(ctx context.Context) bool

+ func (ro Receive[T]) Force(ctx context.Context) T

+ func (ro Receive[T]) Read(ctx context.Context) (T, error)

type Send

+ func (sm Send[T]) Write(ctx context.Context, it T) (err error)

type WorkerFunc

+ func (wf WorkerFunc) Signal(ctx context.Context) <-chan error

May 12, 2023

Changes in this version

+ var ErrSkippedNonBlockingSend = errors.New("skipped non-blocking send")

+ func Count(ctx context.Context, iter Iterator[T]) int

type Iterator

+ func Filter(iter Iterator[T], check func(T) bool) Iterator[T]

+ func PairKeys(iter Iterator[Pair[K, V]]) Iterator[K]

+ func PairValues(iter Iterator[Pair[K, V]]) Iterator[V]

+ type Map map[K]V

+ func Mapify(in map[K]V) Map[K, V]

+ func (m Map[K, V]) Add(k K, v V)

+ func (m Map[K, V]) AddPair(p Pair[K, V])

+ func (m Map[K, V]) Append(pairs ...Pair[K, V])

+ func (m Map[K, V]) Consume(ctx context.Context, iter Iterator[Pair[K, V]])

+ func (m Map[K, V]) ConsumeMap(in Map[K, V])

+ func (m Map[K, V]) ConsumeSlice(in []V, keyf func(V) K)

+ func (m Map[K, V]) ConsumeValues(ctx context.Context, iter Iterator[V], keyf func(V) K)

+ func (m Map[K, V]) Extend(pairs Pairs[K, V])

+ func (m Map[K, V]) Iterator() Iterator[Pair[K, V]]

+ func (m Map[K, V]) Keys() Iterator[K]

+ func (m Map[K, V]) Len() int

+ func (m Map[K, V]) Pairs() Pairs[K, V]

+ func (m Map[K, V]) Values() Iterator[V]

type Pairs

+ func (p *Pairs[K, V]) AddPair(pair Pair[K, V])

+ func (p *Pairs[K, V]) Consume(ctx context.Context, iter Iterator[Pair[K, V]])

+ func (p *Pairs[K, V]) ConsumeMap(in map[K]V)

+ func (p *Pairs[K, V]) ConsumeSlice(in []V, keyf func(V) K)

+ func (p *Pairs[K, V]) ConsumeValues(ctx context.Context, iter Iterator[V], keyf func(V) K)

+ func (p Pairs[K, V]) Iterator() Iterator[Pair[K, V]]

+ func (p Pairs[K, V]) Keys() Iterator[K]

+ func (p Pairs[K, V]) Values() Iterator[V]

type WaitFunc

+ func (wf WaitFunc) Safe() WorkerFunc

type WorkerFunc

+ func (wf WorkerFunc) Add(ctx context.Context, wg *WaitGroup, ob Observer[error])

+ func (wf WorkerFunc) Background(ctx context.Context, ob Observer[error])

May 1, 2023

Changes in this version

+ func Contains(item T, slice []T) bool

Apr 27, 2023

Changes in this version

+ func IterateOneBlocking(iter Iterator[T]) (T, error)

type WaitFunc

+ func (wf WaitFunc) Check() WorkerFunc

Apr 17, 2023

Changes in this version

+ func MustBeOk(out T, ok bool) T

Apr 13, 2023

Changes in this version

+ var ErrSkippedBlockingSend = errors.New("skipped blocking send")

+ func IterateOne(ctx context.Context, iter Iterator[T]) (T, error)

+ func Protect(fn func(I) (O, error)) func(I) (O, error)

+ func WhenCall(cond bool, op func())

+ func WhenDo(cond bool, op func() T) T

type Iterator

+ func Generator(op func(context.Context) (T, error)) Iterator[T]

+ func Transform(iter Iterator[I], op func(in I) (O, error)) Iterator[O]

+ type Observer func(T)

+ func (of Observer[T]) Safe(in T) (err error)

+ func (of Observer[T]) Wait(in T) WaitFunc

+ func (of Observer[T]) Worker(in T) WorkerFunc

+ type Pair struct

+ Key K

+ Value V

+ func MakePair(k K, v V) Pair[K, V]

+ type Pairs []Pair[K, V]

+ func MakePairs(in map[K]V) Pairs[K, V]

+ func (p *Pairs[K, V]) Add(k K, v V)

+ func (p *Pairs[K, V]) UnmarshalJSON(in []byte) error

+ func (p Pairs[K, V]) Append(new ...Pair[K, V]) Pairs[K, V]

+ func (p Pairs[K, V]) Map() map[K]V

+ func (p Pairs[K, V]) MarshalJSON() ([]byte, error)

+ type Send struct

+ func Blocking(ch chan<- T) Send[T]

+ func NonBlocking(ch chan<- T) Send[T]

+ func (sm Send[T]) Check(ctx context.Context, it T) bool

+ func (sm Send[T]) Ignore(ctx context.Context, it T)

+ func (sm Send[T]) Send(ctx context.Context, it T) (err error)

type WorkerFunc

+ func ObserveWorker(iter Iterator[T], fn Observer[T]) WorkerFunc

+ func (wf WorkerFunc) Safe(ctx context.Context) (err error)

Mar 21, 2023

Changes in this version

+ var ErrRecoveredPanic = errors.New("recovered panic")

+ func Ignore(fn func(T) error, arg T)

+ func IgnoreMust(fn func(T) (O, error), arg T) O

Mar 19, 2023

Changes in this version

type WaitFunc

+ func ObserveAll(ctx context.Context, iter Iterator[T], observe func(T)) WaitFunc

+ func ObserveWait(iter Iterator[T], observe func(T)) WaitFunc

type WorkerFunc

+ func ObservePool(ctx context.Context, size int, iter Iterator[T], observe func(T)) WorkerFunc

Mar 18, 2023

Changes in this version

+ func IsWrapped(in T) bool

+ func IsZero(in T) bool

+ func Unwind(in T) []T

+ func Zero(T) T

+ func ZeroOf() T

+ func ZeroWhenNil(val any) T

+ type Atomic struct

+ func NewAtomic(initial T) *Atomic[T]

+ func (a *Atomic[T]) Get() T

+ func (a *Atomic[T]) Set(in T)

type WaitFunc

+ func ObserveWorkerFuncs(ctx context.Context, iter Iterator[WorkerFunc], ob func(error)) WaitFunc

+ func ObserveWorkerPool(ctx context.Context, size int, iter Iterator[WorkerFunc], ob func(error)) WaitFunc

+ func WaitForGroup(wg *sync.WaitGroup) WaitFunc

+ func WaitObserveAllCtx(observe func(context.Context, T), ch <-chan T) WaitFunc

+ func (wf WaitFunc) Add(ctx context.Context, wg *WaitGroup)

+ func (wf WaitFunc) BlockSignal() <-chan struct{}

+ func (wf WaitFunc) CheckWorker() WorkerFunc

+ func (wf WaitFunc) Run(ctx context.Context)

+ func (wf WaitFunc) Signal(ctx context.Context) <-chan struct{}

+ func (wf WaitFunc) WithTimeoutSignal(timeout time.Duration) <-chan struct{}

+ func (wf WaitFunc) Worker() WorkerFunc

type WaitGroup

+ func (wg *WaitGroup) Add(num int)

+ func (wg *WaitGroup) Done()

+ func (wg *WaitGroup) IsDone() bool

+ func (wg *WaitGroup) Num() int

+ func (wg *WaitGroup) Wait(ctx context.Context)

+ type WorkerFunc func(context.Context) error

+ func (wf WorkerFunc) Block() error

+ func (wf WorkerFunc) MustWait() WaitFunc

+ func (wf WorkerFunc) Observe(ctx context.Context, ob func(error))

+ func (wf WorkerFunc) ObserveWait(ob func(error)) WaitFunc

+ func (wf WorkerFunc) Run(ctx context.Context) error

+ func (wf WorkerFunc) Singal(ctx context.Context) <-chan error

+ func (wf WorkerFunc) WithTimeout(timeout time.Duration) error

Mar 3, 2023

Changes in this version

+ func Process(fn func(T), items ...T)

+ func ProcessAll(fn func(T), items []T)

Feb 28, 2023

Changes in this version

+ func Observe(ctx context.Context, iter Iterator[T], observe func(T))

+ func ReadOne(ctx context.Context, ch <-chan T) (T, error)

Feb 28, 2023

Changes in this version

type WaitFunc

+ func (wf WaitFunc) WithTimeout(timeout time.Duration)

Feb 28, 2023

Changes in this version

type WaitFunc

+ func WaitBlocking(fn func()) WaitFunc

+ func WaitBlockingObserve(observe func(T), wait func() T) WaitFunc

Feb 28, 2023

Changes in this version

+ func WaitAdd(ctx context.Context, wg *sync.WaitGroup, fn WaitFunc)

+ type WaitFunc func(context.Context)

+ func WaitChannel(ch <-chan T) WaitFunc

+ func WaitContext(ctx context.Context) WaitFunc

+ func WaitGroup(wg *sync.WaitGroup) WaitFunc

+ func WaitMerge(ctx context.Context, iter Iterator[WaitFunc]) WaitFunc

+ func WaitObserve(observe func(T), ch <-chan T) WaitFunc

+ func WaitObserveAll(observe func(T), ch <-chan T) WaitFunc

+ func (wf WaitFunc) Block()

Feb 23, 2023

Changes in this version

+ func InvariantCheck(fn func() error, args ...any)

+ func InvariantMust(err error, args ...any)

Jan 27, 2023

Changes in this version

+ var ErrInvariantViolation = errors.New("invariant violation")

+ func Check(fn func()) (err error)

+ func Invariant(cond bool, args ...any)

+ func IsInvariantViolation(r any) bool

Jan 21, 2023

Changes in this version

+ func Is(in any) bool

+ func Must(arg T, err error) T

+ func Safe(fn func() T) (out T, err error)

+ func SafeCtx(ctx context.Context, fn func(context.Context) T) (out T, err error)

+ func Unwrap(in T) T

+ func Wait(ctx context.Context, wg *sync.WaitGroup) bool

+ type Iterator interface

+ Close func(context.Context) error

+ Next func(context.Context) bool

+ Value func() T