jobs

Documentation

Index

• Constants
• Variables
• func Parallel(parallel int) func(*Runner)
• type Runner
  • func NewRunner(opts ...RunnerOpt) *Runner
  • func (r *Runner) Run(ctx context.Context, cancel context.CancelCauseFunc, workQueue WorkQueue) *sync.WaitGroup
• type RunnerOpt
• type Work
• type WorkQueue
  • func Filter[T any](buffer int, in <-chan T, predicate func(T) (bool, error)) (<-chan T, WorkQueue)
  • func ForEach[FROM any](buffer int, in <-chan FROM, fn func(FROM) error) WorkQueue
  • func Map[FROM, TO any](buffer int, in <-chan FROM, fn func(FROM) (TO, error)) (<-chan TO, WorkQueue)
  • func Reduce[T any](buffer int, in <-chan T, fn func(T) error) WorkQueue

Constants

const WorkBuffer = 100

Variables

var ErrStream = errors.New("executing workflow")

Functions

func Parallel

func Parallel(parallel int) func(*Runner)

Types

type Runner

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

func NewRunner

func NewRunner(opts ...RunnerOpt) *Runner

func (*Runner) Run

func (r *Runner) Run(ctx context.Context, cancel context.CancelCauseFunc, workQueue WorkQueue) *sync.WaitGroup

Run launches goroutines that execute work in workQueue. Returns a WaitGroup that completes when all work is done. Halts and marks WaitGroup as done on error.

type RunnerOpt

type RunnerOpt func(*Runner)

type Work

type Work func() error

type WorkQueue

type WorkQueue <-chan Work

func Filter

func Filter[T any](buffer int, in <-chan T, predicate func(T) (bool, error)) (<-chan T, WorkQueue)

Filter discards elements which do not cause predicate to return true.

func ForEach

func ForEach[FROM any](buffer int, in <-chan FROM, fn func(FROM) error) WorkQueue

ForEach performs an action for each item, returning nothing.

func Map

func Map[FROM, TO any](buffer int, in <-chan FROM, fn func(FROM) (TO, error)) (<-chan TO, WorkQueue)

Map mutates an input channel, optionally changing the type.

func Reduce

func Reduce[T any](buffer int, in <-chan T, fn func(T) error) WorkQueue

Reduce consumes a channel.