async

type CancelContext ¶

type CancelContext = context.CancelContext

CancelContext is a cancellable async context.

func CancelledContext() CancelContext

func NewContext() CancelContext

func NextContext(parent Context) CancelContext

type Context ¶

type Context = context.Context

Context is an async cancellation context.

func DeadlineContext(deadline time.Time) Context

func NextDeadlineContext(parent Context, deadline time.Time) Context

func NextTimeoutContext(parent Context, timeout time.Duration) Context

func NoContext() Context

func TimeoutContext(timeout time.Duration) Context

type ContextCallback ¶

type ContextCallback = context.Callback

ContextCallback is called when the context is cancelled.

type Flag ¶

type Flag = flag.Flag

Flag is a read-only boolean flag that can be waited on until set.

Example:

serving := async.UnsetFlag()

func handle(ctx Context, req *request) {
	if !serving.Get() { // just to show Get in example
		select {
		case <-ctx.Wait():
			return ctx.Status()
		case <-serving.Wait():
		}
	}

	// ... handle request
}

func ReverseFlag(f Flag) Flag

type Func ¶

type Func[T any] func(ctx Context) (T, status.Status)

Func is a function which returns the result.

type Func1 ¶

type Func1[T any, A any] func(ctx Context, arg A) (T, status.Status)

Func1 is a single argument function which returns the result.

type FuncVoid ¶

type FuncVoid = func(ctx Context) status.Status

FuncVoid is a function which returns no result.

type FuncVoid1 ¶

type FuncVoid1[A any] func(ctx Context, arg A) status.Status

FuncVoid1 is a single argument function which returns no result

type Future ¶

type Future[T any] interface {
	// Done returns true if the future is complete.
	Done() bool

	// Wait returns a channel which is closed when the future is complete.
	Wait() <-chan struct{}

	// Result returns a value and a status.
	Result() (T, status.Status)

	// Status returns a status or none.
	Status() status.Status
}

Future represents a result available in the future.

func Completed[T any](result T, st status.Status) Future[T]

func Rejected[T any](st status.Status) Future[T]

func Resolved[T any](result T) Future[T]

type FutureDyn ¶

type FutureDyn interface {
	// Done returns true if the future is complete.
	Done() bool

	// Wait returns a channel which is closed when the future is complete.
	Wait() <-chan struct{}

	// Status returns a status or none.
	Status() status.Status
}

FutureDyn is a future interface without generics, i.e. Future[?].

type FutureGroup ¶

type FutureGroup[T any] []Future[T]

FutureGroup is a group of futures of the same type. Use FutureGroupDyn for a group of futures of different types.

func (g FutureGroup[T]) Await(ctx context.Context) status.Status

func (g FutureGroup[T]) AwaitAny(ctx context.Context) (T, int, status.Status)

func (g FutureGroup[T]) AwaitError(ctx context.Context) (int, status.Status)

func (g FutureGroup[T]) AwaitResults(ctx context.Context) ([]Result[T], status.Status)

func (g FutureGroup[T]) Results() []Result[T]

func (g FutureGroup[T]) Statuses() []status.Status

type FutureGroupDyn ¶

type FutureGroupDyn []FutureDyn

FutureGroupDyn is a group of futures of different types. Use FutureGroup for a group of futures of the same type.

func (g FutureGroupDyn) Await(ctx context.Context) status.Status

func (g FutureGroupDyn) AwaitAny(ctx context.Context) (int, status.Status)

func (g FutureGroupDyn) AwaitError(ctx context.Context) (int, status.Status)

func (g FutureGroupDyn) Statuses() []status.Status

type Lock ¶

type Lock = lock.Lock

Lock is a channel-based lock, which be used directly as a channel, or via the Lock/Unlock methods. To lock a lock receive an empty struct from it, to unlock a lock send an empty struct to it.

Example:

lock := async.NewLock()
select {
case <-lock:
case <-cancel:
	return status.Cancelled
}
defer lock.Unlock()

func NewLock() Lock

type MutFlag ¶

type MutFlag = flag.MutFlag

MutFlag is a routine-safe boolean flag that can be set, reset, and waited on until set.

Example:

serving := async.UnsetFlag()

func serve() {
	s.serving.Set()
	defer s.serving.Unset()

	// ... start server ...
}

func handle(ctx Context, req *request) {
	select {
	case <-ctx.Wait():
		return ctx.Status()
	case <-serving.Wait():
	}

	// ... handle request
}

func SetFlag() MutFlag

func UnsetFlag() MutFlag

type NextFunc ¶

type NextFunc[T any] func(ctx context.Context) (T, bool, status.Status)

NextFunc is a function that returns the next value from a stream.

type Pool ¶

type Pool = pool.Pool

Pool allows to reuse goroutines with preallocated big stacks.

func NewPool() Pool

type Promise ¶

type Promise[T any] interface {
	Future[T]

	// Reject rejects the promise with a status.
	Reject(st status.Status) bool

	// Resolve completes the promise with a result and ok.
	Resolve(result T) bool

	// Complete completes the promise with a status and a result.
	Complete(result T, st status.Status) bool
}

Promise is a future which can be completed.

func NewPromise[T any]() Promise[T]

type Queue ¶

type Queue[T any] interface {
	// Len returns the number of elements in the queue.
	Len() int

	// Clear clears the queue.
	Clear()

	// Push adds an element to the queue.
	Push(v T)

	// Poll removes an element from the queue, returns false if the queue is empty.
	Poll() (T, bool)

	// Wait returns a channel which is notified on new elements.
	Wait() <-chan struct{}
}

Queue is an unbounded FIFO queue guarded by a mutex.

func NewQueue[T any](items ...T) Queue[T]

type Result ¶

type Result[T any] struct {
	Value  T
	Status status.Status
}

Result is an async result which combines a value and a status.

func AwaitResults[F Future[T], T any](ctx context.Context, futures ...F) ([]Result[T], status.Status)

func (r Result[T]) Unwrap() (T, status.Status)

type Routine ¶

type Routine[T any] interface {
	Future[T]

	// Start start the routine, if not started or stopped yet.
	Start()

	// Stop requests the routine to stop and returns a wait channel.
	// The method does not call the on-stop callbacks if the routine has not started.
	Stop() <-chan struct{}

	// OnStop adds a callback which is called when the routine stops.
	// The callback is called by the routine goroutine, and is not called
	// if the routine has not started.
	OnStop(fn func(Routine[T])) bool
}

Routine is an async routine which returns the result as a future, recovers on panics, and can be cancelled.

func NewRoutine[T any](fn Func[T]) Routine[T]

func Run[T any](fn Func[T]) Routine[T]

func Run1[T any, A any](fn Func1[T, A], arg A) Routine[T]

func Stopped[T any](result T, st status.Status) Routine[T]

type RoutineDyn ¶

type RoutineDyn interface {
	FutureDyn

	// Start starts the routine, if not stopped already.
	Start()

	// Stop requests the routine to stop and returns a wait channel.
	Stop() <-chan struct{}
}

RoutineDyn is a routine interface without generics, i.e. Routine[?].

type RoutineGroup ¶

type RoutineGroup[T any] []Routine[T]

RoutineGroup is a group of routines of the same type. Use RoutineGroupDyn for a group of routines of different types.

func (g RoutineGroup[T]) Await(ctx Context) status.Status

func (g RoutineGroup[T]) AwaitAny(ctx Context) (T, int, status.Status)

func (g RoutineGroup[T]) AwaitError(ctx Context) (int, status.Status)

func (g RoutineGroup[T]) AwaitResults(ctx Context) ([]Result[T], status.Status)

func (g RoutineGroup[T]) Results() []Result[T]

func (g RoutineGroup[T]) Statuses() []status.Status

func (g RoutineGroup[T]) Stop()

func (g RoutineGroup[T]) StopWait()

func (g RoutineGroup[T]) Values() []T

type RoutineGroupDyn ¶

type RoutineGroupDyn []RoutineDyn

RoutineGroupDyn is a group of routines of different types. Use RoutineGroup for a group of routines of the same type.

func (g RoutineGroupDyn) Await(ctx Context) status.Status

func (g RoutineGroupDyn) AwaitAny(ctx Context) (int, status.Status)

func (g RoutineGroupDyn) AwaitError(ctx Context) (int, status.Status)

func (g RoutineGroupDyn) Statuses() []status.Status

func (g RoutineGroupDyn) Stop()

func (g RoutineGroupDyn) StopWait()

type RoutineVoid ¶

type RoutineVoid = Routine[struct{}]

RoutineVoid is a routine which has no result.

func NewRoutineVoid(fn func(ctx Context) status.Status) RoutineVoid

func RunVoid(fn FuncVoid) RoutineVoid

func RunVoid1[A any](fn FuncVoid1[A], arg A) RoutineVoid

func StoppedVoid(st status.Status) RoutineVoid

type Runner ¶

type Runner interface {
	// Run runs the function.
	Run()
}

Runner is an interface that provides a run method.

func RunnerFunc(fn func()) Runner

type Service ¶

type Service interface {
	// Status returns the stop status or none.
	Status() status.Status

	// Running indicates that the service routine is running.
	Running() Flag

	// Stopped indicates that the service is stopped.
	Stopped() Flag

	// Start starts the service if not running.
	Start() status.Status

	// Stop requests the service to stop and returns its routine or a stopped routine.
	Stop() <-chan struct{}

	// Wait returns a channel which is closed when the service is stopped.
	Wait() <-chan struct{}
}

Service is a service which can be started and stopped.

func NewService(fn func(ctx Context) status.Status) Service

type StopGroup ¶

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

StopGroup stops all operations in the group and awaits their completion.

func NewStopGroup() *StopGroup

func (g *StopGroup) Add(s Stopper)

func (g *StopGroup) AddMany(stoppers ...Stopper)

func (g *StopGroup) AddService(s Service)

func (g *StopGroup) AddTicker(t *time.Ticker)

func (g *StopGroup) AddTimer(t *time.Timer)

func (g *StopGroup) Stop()

func (g *StopGroup) StopWait()

func (g *StopGroup) Wait()

type Stopper ¶

type Stopper interface {
	// Wait returns a channel which is closed when the future is complete.
	Wait() <-chan struct{}

	// Stop requests the routine to stop and returns a wait channel.
	Stop() <-chan struct{}
}

Stopper stops a routine and awaits its stop.

type Stream ¶

type Stream[T any] interface {
	// Next returns the next value from the stream, or false if the stream has ended.
	Next(ctx context.Context) (T, bool, status.Status)

	// Free frees the stream.
	Free()
}

Stream is an async stream of values.

func NewStream[T any](next NextFunc[T]) Stream[T]

func NewStreamFree[T any](next NextFunc[T], free func()) Stream[T]

func NewStreamFreer[T any](next NextFunc[T], freer ref.Freer) Stream[T]

type Variable ¶

type Variable[T any] interface {
	// Get returns the current value/error, or false if pending.
	Get() (T, bool, status.Status)

	// GetWait returns the current value or waits for it or an error.
	GetWait(ctx context.Context) (T, status.Status)

	// Clear clears the variable.
	Clear()

	// Complete sets the value and status.
	Complete(value T, st status.Status)

	// Fail sets the error.
	Fail(st status.Status)

	// Set sets the value.
	Set(value T)

	// Wait returns a channel that is closed when the variable is set or failed.
	Wait() <-chan struct{}
}

[Experimental] Variable is an asynchronous variable which can be set, cleared, or failed.

func NewVariable[T any]() Variable[T]

type WaitLock ¶

type WaitLock = lock.WaitLock

WaitLock is a lock which allows others to wait until it is unlocked.

WaitLock does not guarantee that the lock is not acquired by another writer after its waiters are notified.

WaitLock can be used, for example, to execute a single operation by one writer, while other writers wait until the operation is completed.

Example:

lock := async.NewWaitLock()

func flush(cancel <-chan struct{}) {
	select {
	case <-lock.Lock():
		// Acquired lock
	default:
		// Await flushing end
		select {
		case <-lock.Wait():
		case <-cancel:
			return status.Cancelled
		}
	}
	defer lock.Unlock()

	// ... Do work ...
}

func NewWaitLock() WaitLock