fun

README

fun -- Go Generic Functions and Tools

fun is a simple, well tested, zero-dependency, collection of packages with generic function, tools, patterns, and the kind of thing you could write one-offs for but shouldn't.

Packages:

• erc (error collecting)
• itertool (iterator tools)
• pubsub (message broker and queues)
• set (generic ordered and unordered sets)
• seq (generic linked lists.)
• srv (service orchestration and management framework.)

For more information, see the documentation, but of general interest:

• In itertools and with fun.Iterator, an iterator framework and tools for interacting with iterators and generators.
• In pubsub, a channel-based message broker (for one-to-many channel patterns), with several backend patterns for dealing with load-shedding and message distribution patterns.
• In erc, an error collector implementation for threadsafe error aggregation and introspection, particularly in worker-pool, applications.
• In set, a Set type, with ordered and unordered implementations.
• Queue and Deque implementations (in pubsub) that provide thread-safe linked-list based implementations and Wait methods to block until new items added.
• In seq, general purpose linked list implementations, with a healthy feature set and flexible interface.
• In srv, a service orchestration toolkit.

Contributions welcome, the general goals of the project:

• superior API ergonomics.
• great high-level abstractions.
• obvious and clear implementations.
• minimal dependencies.

Have fun!

Documentation

Overview

Package fun is a zero-dependency collection of tools and idoms that takes advantage of generics. Iterators, error handling, a native-feeling Set type, and a simple pub-sub framework for distributing messages in fan-out patterns.

Index

• Variables
• func Check(fn func()) (err error)
• func Invariant(cond bool, args ...any)
• func InvariantCheck(fn func() error, args ...any)
• func InvariantMust(err error, args ...any)
• func Is[T any](in any) bool
• func IsInvariantViolation(r any) bool
• func IsWrapped[T any](in T) bool
• func IsZero[T comparable](in T) bool
• func Must[T any](arg T, err error) T
• func Observe[T any](ctx context.Context, iter Iterator[T], observe func(T))
• func ReadOne[T any](ctx context.Context, ch <-chan T) (T, error)
• func Safe[T any](fn func() T) (out T, err error)
• func Unwind[T any](in T) []T
• func Unwrap[T any](in T) T
• func Zero[T any](T) T
• func ZeroOf[T any]() T
• func ZeroWhenNil[T any](val any) T
• type Atomic
  • func NewAtomic[T any](initial T) *Atomic[T]
  • func (a *Atomic[T]) Get() T
  • func (a *Atomic[T]) Set(in T)
• type Iterator
• 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 WaitBlocking(fn func()) WaitFunc
  • func WaitBlockingObserve[T any](observe func(T), wait func() T) WaitFunc
  • func WaitChannel[T any](ch <-chan T) WaitFunc
  • func WaitContext(ctx context.Context) WaitFunc
  • func WaitForGroup(wg *sync.WaitGroup) WaitFunc
  • func WaitMerge(ctx context.Context, iter Iterator[WaitFunc]) WaitFunc
  • func WaitObserve[T any](observe func(T), ch <-chan T) WaitFunc
  • func WaitObserveAll[T any](observe func(T), ch <-chan T) WaitFunc
  • func WaitObserveAllCtx[T any](observe func(context.Context, T), ch <-chan T) WaitFunc
  • func (wf WaitFunc) Add(ctx context.Context, wg *WaitGroup)
  • func (wf WaitFunc) Block()
  • 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) WithTimeout(timeout time.Duration)
  • 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 (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

Variables

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

ErrInvariantViolation is the root error of the error object that is the content of all panics produced by the Invariant helper.

Functions

func Check

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

Check, like Safe, runs a function without arguments that does not produce an error, and, if the function panics, converts it into an error.

func Invariant

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

Invariant panics if the condition is false Invariant panics, passing an error that is rooted by ErrInvariantViolation.

func InvariantCheck

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

InvariantCheck calls the function and if it returns an error panics with an ErrInvariantViolation error, wrapped with the error of the function, and any annotation arguments.

func InvariantMust

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

InvariantMust raises an invariant error if the error is not nil. The content of the panic is both--via wrapping--an ErrInvariantViolation and the error itself.

func Is

func Is[T any](in any) bool

Is a generic version of `errors.Is` that takes advantage of the Unwrap function, and is useful for checking if an object of an interface type is or wraps an implementation of the type parameter.

func IsInvariantViolation

func IsInvariantViolation(r any) bool

IsInvariantViolation returns true if the argument is or resolves to ErrInvariantViolation.

func IsWrapped

func IsWrapped[T any](in T) bool

IsWrapped returns true if the input value wraps another value of the same type.

func IsZero

func IsZero[T comparable](in T) bool

IsZero returns true if the input value compares "true" to the zero value for the type of the argument. If the type implements an IsZero() method (e.g. time.Time), then IsZero returns that value, otherwise, IsZero constructs a zero valued object of type T and compares the input value to the zero value.

func Must

func Must[T any](arg T, err error) T

Must wraps a function that returns a value and an error, and converts the error to a panic.

func Observe

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

Observe processes an iterator calling the observer function for every element in the iterator and retruning when the iterator is exhausted. Take care to ensure that the Observe function does not block.

Use itertool.Observe and itertool.ParallelObserve for more advanced execution patterns.

Use with itertool.Slice, itertool.Channel, or itertool.Variadic to process data in other forms.

func ReadOne

func ReadOne[T any](ctx context.Context, ch <-chan T) (T, error)

ReadOnce reads one item from the channel, and returns it. ReadOne returns early if the context is canceled (ctx.Err()) or the channel is closed (io.EOF).

func Safe

func Safe[T any](fn func() T) (out T, err error)

Safe runs a function with a panic handler that converts the panic to an error.

func Unwind

func Unwind[T any](in T) []T

Unwind uses the Unwrap operation to build a list of the "wrapped" objects.

func Unwrap

func Unwrap[T any](in T) T

Unwrap is a generic equivalent of the `errors.Unwrap()` function for any type that implements an `Unwrap() T` method. useful in combination with Is.

func Zero

func Zero[T any](T) T

Zero returns the zero-value for the type T of the input argument.

func ZeroOf

func ZeroOf[T any]() T

ZeroOf returns the zero-value for the type T specified as an argument.

func ZeroWhenNil

func ZeroWhenNil[T any](val any) T

ZeroWhenNil takes a value of any type, and if that value is nil, returns the zero value of the specified type. Otherwise, ZeroWhenNil coerces the value into T and returns it. If the input value does not match the output type of the function, ZeroWhenNil panics with an ErrInvariantViolation.

Types

type Atomic

type Atomic[T any] struct{ atomic.Value }

Atomic is a very simple atomic Get/Set operation, providing a generic type-safe implementation wrapping sync/atomic.Value.

func NewAtomic

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

NewAtomic creates a new Atomic Get/Set value with the initial value already set.

func (*Atomic[T]) Get

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

Get resolves the atomic value, returning the zero value of the type T if the value is unset.

func (*Atomic[T]) Set

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

Set atomically sets the value of the Atomic.

type Iterator

type Iterator[T any] interface {
	Next(context.Context) bool
	Close() error
	Value() T
}

Iterator provides a safe, context-respecting iterator paradigm for iterable objects, along with a set of consumer functions and basic implementations.

The itertool package provides a number of tools and paradigms for creating and processing Iterator objects, including Generators, Map and Reduce, Filter as well as Split and Merge to combine or divide iterators.

In general, Iterators cannot be safe for access from multiple concurrent goroutines, because it is impossible to synchronize calls to Next() and Value(); however, itertool.Range() and itertool.Split() provide support for these workloads.

type WaitFunc

type WaitFunc func(context.Context)

WaitFunc is a type of function object that will block until an operation returns or the context is canceled.

func ObserveWorkerFuncs

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

ObserveWorkerFuncs runs each worker function present in the iterator running until the iterator stops provding values. This is a non-blocking function, and the returned waitfunction blocks until all Workers *and* the iterator have completed.

Only non-nil error outputs from the worker functions are passed to the observer. Panics are not handled. This worker pool is unbounded.

Use itertool.Channel, itertool.Slice, and itertool.Variadic to produce iterators from standard types for use with this function.

func ObserveWorkerPool

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

ObserveWorkerPool creates a worker pool of the specified size and dispatches worker functions from the iterator until the iterator is closed and all worker functions have returned. The context passed to the worker functions is always a decedent of the context passed to ObserveWorkerPool.

This operation is non-blocking. Callers must call the WaitFunc to ensure all work is complete.

Only non-nil error outputs from the worker functions are passed to the observer. Panics in workers are not handled. If the size is <= 0 then a size of 1 is used.

Use itertool.Channel, itertool.Slice, and itertool.Variadic to produce iterators from standard types for use with this function.

func WaitBlocking

func WaitBlocking(fn func()) WaitFunc

WaitBlocking is a convenience function to use simple blocking functions into WaitFunc objects. Because these WaitFunc functions do not resepct the WaitFunc context, use with care and caution.

func WaitBlockingObserve

func WaitBlockingObserve[T any](observe func(T), wait func() T) WaitFunc

WaitBlockingObserve is a convenience function that creates a WaitFunc that wraps a simple function that returns a single value, and observes that output with the observer function.

Because these WaitFunc functions do not resepct the WaitFunc context, use with care and caution.

func WaitChannel

func WaitChannel[T any](ch <-chan T) WaitFunc

WaitChannel converts a channel (typically, a `chan struct{}`) to a WaitFunc. The WaitFunc blocks till it's context is canceled or the channel is either closed or returns one item.

func WaitContext

func WaitContext(ctx context.Context) WaitFunc

WaitContext wait's for the context to be canceled before returning. The WaitFunc that's return also respects it's own context. Use this WaitFunc and it's own context to wait for a context to be cacneled with a timeout, for instance.

func WaitForGroup

func WaitForGroup(wg *sync.WaitGroup) WaitFunc

WaitForGroup converts a sync.WaitGroup into a fun.WaitFunc.

This operation will leak a go routine if the WaitGroup never returns and the context is canceled. To avoid a leaked goroutine, use the fun.WaitGroup type.

func WaitMerge

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

WaitMerge starts a goroutine that blocks on each WaitFunc provided and returns a WaitFunc that waits for all of these goroutines to return. The constituent WaitFunc are passed WaitMerge's context, while the returned WaitFunc respects its own context.

Use itertool.Variadic, itertool.Slice, or itertool.Channel to convert common container types/calling patterns to an iterator.

In combination with erc.CheckWait, you can use WaitMerge to create and pubsub.Queue or pubsub.Deque blocking iterators to create worker pools.

func WaitObserve

func WaitObserve[T any](observe func(T), ch <-chan T) WaitFunc

WaitObserve passes the output of the channel into the observer function and then returns. If the context is canceled the output of the channel is not observed.

WaitObserve consumes and observes, at most, one item from the channel. Callers must call the WaitFunc.

func WaitObserveAll

func WaitObserveAll[T any](observe func(T), ch <-chan T) WaitFunc

WaitObserveAll passes the output of the channel into the observer function, waiting for the input channel to be closed or the WaitFunc's context to be canceled. WaitObserveAll does not begin processing the channel until the WaitFunc is called.

func WaitObserveAllCtx

func WaitObserveAllCtx[T any](observe func(context.Context, T), ch <-chan T) WaitFunc

WaitObserveAllCtx passes the output of the channel into the observer function with a context, waiting for the input channel to be closed or the WaitFunc's context to be canceled. WaitObserveAll does not begin processing the channel until the WaitFunc is called.

func (WaitFunc) Add

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

Add starts a goroutine that waits for the WaitFunc to return, incrementing and decrementing the sync.WaitGroup as appropriate. The execution of the wait fun blocks on Add's context.

func (WaitFunc) Block

func (wf WaitFunc) Block()

Block runs the WaitFunc with a context that will never be canceled.

func (WaitFunc) BlockSignal

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

BlockSignal runs the WaitFunc in a background goroutine and returns a signal channel that is closed when the operation completes. As in Block() the WaitFunc is passed a background context that is never canceled.

Callers are responsble for handling the (potential) panic in the WaitFunc.

func (WaitFunc) CheckWorker

func (wf WaitFunc) CheckWorker() WorkerFunc

CheckWorker converts a wait function into a WorkerFunc, running the wait function inside of a Check() function, which catches panics and turns them into the worker function's errors.

func (WaitFunc) Run

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

Run is equivalent to calling the wait function directly, except the context passed to the function is always canceled when the wait function returns.

func (WaitFunc) Signal

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

Signal runs the WaitFunc in a goroutine and returns a signal channel that is canceled when the function completes. Useful for bridging the gap between interfaces and integration that use channels and functions.

Callers are responsble for handling the (potential) panic in the WaitFunc.

func (WaitFunc) WithTimeout

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

WithTimeout runs the WaitFunc with an explicit timeout.

func (WaitFunc) WithTimeoutSignal

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

WithTimeoutSignal executes the WaitFunc as in WithTimeout, but returns a singal channel that is closed when the task completes.

Callers are responsble for handling the (potential) panic in the WaitFunc.

func (WaitFunc) Worker

func (wf WaitFunc) Worker() WorkerFunc

Worker converts a wait function into a WorkerFunc. These workers will never error.

type WaitGroup

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

WaitGroup works like sync.WaitGroup, except that the Wait method takes a context (and can be passed as a fun.WaitFunc). The implementation is exceptionally simple. The only constraint is that you can never modify the value of the internal counter such that it is negative, event transiently. The implementation does not require background resources aside from Wait, which creates a goroutine that lives for the entire time that Wait is running. Multiple copies of Wait can be safely called at once, and the WaitGroup is reusable more than once.

func (*WaitGroup) Add

func (wg *WaitGroup) Add(num int)

Add modifies the internal counter. Raises an ErrInvariantViolation error if any modification causes the internal coutner to be less than 0.

func (*WaitGroup) Done

func (wg *WaitGroup) Done()

Done marks a single operation as done.

func (*WaitGroup) IsDone

func (wg *WaitGroup) IsDone() bool

IsDone returns true if there is pending work, and false otherwise.

func (*WaitGroup) Num

func (wg *WaitGroup) Num() int

Num returns the number of pending workers.

func (*WaitGroup) Wait

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

Wait blocks until either the context is canceled or the

type WorkerFunc

type WorkerFunc func(context.Context) error

WorkerFunc represents a basic function used in worker pools and other similar situations

func (WorkerFunc) Block

func (wf WorkerFunc) Block() error

Block executes the worker function with a context that will never expire and returns the error. Use with caution

func (WorkerFunc) MustWait

func (wf WorkerFunc) MustWait() WaitFunc

MustWait converts a Worker function into a wait function; however, if the worker produces an error MustWait converts the error into a panic.

func (WorkerFunc) Observe

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

Observe runs the worker function passing the error output to the observer function. Only non-nil errors are observed.

func (WorkerFunc) ObserveWait

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

ObserveWait converts a worker function into a wait function, passing any error to the observer function. Only non-nil errors are observed.

func (WorkerFunc) Run

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

Run is equivalent to calling the worker function directly, except the context passed to it is canceled when the worker function returns.

func (WorkerFunc) Singal

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

Signal runs the worker function in a background goroutine and returns the error in an error channel, that returns when the worker function returns. If Singal is called with a canceled context the worker is still executed (with that context.)

func (WorkerFunc) WithTimeout

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

WithTimeout executes the worker function with the provided timeout using a new context.