async

Documentation

Index

• func ForwardContext[T any](ctx context.Context, dst chan<- T, src <-chan T)
• func GoAnnotated(ctx context.Context, panicHandler PanicHandler, fn func(context.Context), ...)
• func HandlePanic(panicHandler PanicHandler)
• func RangeContext[T any](ctx context.Context, ch <-chan T, fn func(T))
• type Abortable
  • func (a *Abortable) Abort()
  • func (a *Abortable) Do(ctx context.Context, fn func(context.Context))
• type Group
  • func NewGroup(ctx context.Context, panicHandler PanicHandler) *Group
  • func (g *Group) Cancel()
  • func (g *Group) CancelAndWait()
  • func (g *Group) Finish()
  • func (g *Group) Once(f func(ctx context.Context))
  • func (g *Group) Periodic(interval time.Duration, jitter time.Duration, f func(ctx context.Context))
  • func (g *Group) PeriodicOrTrigger(interval time.Duration, jitter time.Duration, f func(ctx context.Context)) func()
  • func (g *Group) Trigger(f func(ctx context.Context)) func()
  • func (g *Group) WaitToFinish()
• type NoopPanicHandler
  • func (n NoopPanicHandler) HandlePanic(r interface{})
• type PanicHandler
• type QueuedChannel
  • func NewQueuedChannel[T any](chanBufferSize, queueCapacity int, panicHandler PanicHandler) *QueuedChannel[T]
  • func (q *QueuedChannel[T]) Close()
  • func (q *QueuedChannel[T]) CloseAndDiscardQueued()
  • func (q *QueuedChannel[T]) Enqueue(items ...T) bool
  • func (q *QueuedChannel[T]) GetChannel() <-chan T
• type WaitGroup
  • func MakeWaitGroup(panicHandler PanicHandler) WaitGroup
  • func (wg *WaitGroup) Go(f func())
  • func (wg *WaitGroup) Wait()

Functions

func ForwardContext

func ForwardContext[T any](ctx context.Context, dst chan<- T, src <-chan T)

ForwardContext forwards all values from the src channel to the dst channel until the context is canceled or the src channel is closed.

func GoAnnotated

func GoAnnotated(ctx context.Context, panicHandler PanicHandler, fn func(context.Context), labelMap ...logging.Labels)

func HandlePanic

func HandlePanic(panicHandler PanicHandler)

func RangeContext

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

RangeContext iterates over the given channel until the context is canceled or the channel is closed.

Types

type Abortable

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

Abortable collects groups of functions that can be aborted by calling Abort.

func (*Abortable) Abort

func (a *Abortable) Abort()

func (*Abortable) Do

func (a *Abortable) Do(ctx context.Context, fn func(context.Context))

type Group

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

Group is forked and improved version of "github.com/bradenaw/juniper/xsync.Group".

It manages a group of goroutines. The main change to original is posibility to wait passed function to finish without canceling it's context and adding PanicHandler.

func NewGroup

func NewGroup(ctx context.Context, panicHandler PanicHandler) *Group

NewGroup returns a Group ready for use. The context passed to any of the f functions will be a descendant of ctx.

func (*Group) Cancel

func (g *Group) Cancel()

Cancel is send to all of the spawn goroutines and ends periodic or trigger routines.

func (*Group) CancelAndWait

func (g *Group) CancelAndWait()

CancelAndWait cancels the context passed to any of the spawned goroutines and waits for all spawned goroutines to exit.

It is not safe to call Wait concurrently with any other method on g.

func (*Group) Finish

func (g *Group) Finish()

Finish will ends all periodic or polls routines. It will let currently running functions to finish (cancel is not sent).

It is not safe to call Wait concurrently with any other method on g.

func (*Group) Once

func (g *Group) Once(f func(ctx context.Context))

Once calls f once from another goroutine.

func (*Group) Periodic

func (g *Group) Periodic(
	interval time.Duration,
	jitter time.Duration,
	f func(ctx context.Context),
)

Periodic spawns a goroutine that calls f once per interval +/- jitter.

func (*Group) PeriodicOrTrigger

func (g *Group) PeriodicOrTrigger(
	interval time.Duration,
	jitter time.Duration,
	f func(ctx context.Context),
) func()

PeriodicOrTrigger spawns a goroutine which calls f whenever the returned function is called. If f is already running when triggered, f will run again immediately when it finishes. Also calls f when it has been interval+/-jitter since the last trigger.

func (*Group) Trigger

func (g *Group) Trigger(f func(ctx context.Context)) func()

Trigger spawns a goroutine which calls f whenever the returned function is called. If f is already running when triggered, f will run again immediately when it finishes.

func (*Group) WaitToFinish

func (g *Group) WaitToFinish()

WaitToFinish will ends all periodic or polls routines. It will wait for currently running functions to finish (cancel is not sent).

It is not safe to call Wait concurrently with any other method on g.

type NoopPanicHandler

type NoopPanicHandler struct{}

func (NoopPanicHandler) HandlePanic

func (n NoopPanicHandler) HandlePanic(r interface{})

type PanicHandler

type PanicHandler interface {
	HandlePanic(interface{})
}

type QueuedChannel

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

QueuedChannel represents a channel on which queued items can be published without having to worry if the reader has actually consumed existing items first or if there's no way of knowing ahead of time what the ideal channel buffer size should be.

func NewQueuedChannel

func NewQueuedChannel[T any](chanBufferSize, queueCapacity int, panicHandler PanicHandler) *QueuedChannel[T]

func (*QueuedChannel[T]) Close

func (q *QueuedChannel[T]) Close()

func (*QueuedChannel[T]) CloseAndDiscardQueued

func (q *QueuedChannel[T]) CloseAndDiscardQueued()

CloseAndDiscardQueued force closes the channel and does not guarantee that the remaining queued items will be read.

func (*QueuedChannel[T]) Enqueue

func (q *QueuedChannel[T]) Enqueue(items ...T) bool

func (*QueuedChannel[T]) GetChannel

func (q *QueuedChannel[T]) GetChannel() <-chan T

type WaitGroup

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

func MakeWaitGroup

func MakeWaitGroup(panicHandler PanicHandler) WaitGroup

func (*WaitGroup) Go

func (wg *WaitGroup) Go(f func())

func (*WaitGroup) Wait

func (wg *WaitGroup) Wait()